Limess
commented
3 months ago After upgrading to 0.14.1 (EMR 7.1.0) this is still occuring. This didn't happen until enabling metadata at which point the timeline server becomes a bottleneck and source of failures.
This only occurs during our "backfill" job where we rewrite most of the table, our incremental loads with small clusters don't exhibit this issue.
Any recommendations to mitigate this? The cluster is pretty large using this config:
- Is the driver too small for the cluster size?
- Any way we can verify that using logs/metrics?
- Would altering
hoodie.embed.timeline.server.threadshelp - it seems this is fixed to 200 threads regardless of anything else, despite the documentation suggesting it's variable. - Would enabling
hoodie.embed.timeline.server.asynchelp? - Anything else we can tweak?
From what I can tell based on OS level metrics, the driver node (r6g.4xlarge, driver + 3 executors) is barely doing anything CPU wise? It seems to be using < 10% CPU
[
{
"Classification": "spark",
"Properties": {
"maximizeResourceAllocation": "false"
}
},
{
"Classification": "spark-defaults",
"Properties": {
"spark.default.parallelism": "3352",
"spark.driver.cores": "4",
"spark.driver.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35",
"spark.driver.memory": "25g",
"spark.driver.memoryOverhead": "3g",
"spark.dynamicAllocation.enabled": "false",
"spark.executor.cores": "4",
"spark.executor.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35",
"spark.executor.instances": "419",
"spark.executor.maxNumFailures": "100",
"spark.executor.memory": "25g",
"spark.executor.memoryOverhead": "3g",
"spark.executor.processTreeMetrics.enabled": "true",
"spark.executorEnv.PEX_INHERIT_PATH": "fallback",
"spark.hadoop.fs.s3.connection.maximum": "1000",
"spark.hadoop.fs.s3a.connection.maximum": "1000",
"spark.kryoserializer.buffer.max": "256m",
"spark.metrics.namespace": "spark",
"spark.rdd.compress": "true",
"spark.scheduler.mode": "FAIR",
"spark.serializer": "org.apache.spark.serializer.KryoSerializer",
"spark.shuffle.service.enabled": "true",
"spark.sql.adaptive.coalescePartitions.enabled": "true",
"spark.sql.shuffle.partitions": "3352",
"spark.task.maxFailures": "10",
"spark.ui.prometheus.enabled": "true",
"spark.yarn.appMasterEnv.PEX_INHERIT_PATH": "fallback",
"spark.yarn.maxAppAttempts": "1"
}
},
{
"Classification": "spark-log4j2",
"Properties": {
"logger.hudi.level": "INFO",
"logger.hudi.name": "org.apache.hudi"
}
},
{
"Classification": "spark-metrics",
"Properties": {
"*.sink.prometheusServlet.class": "org.apache.spark.metrics.sink.PrometheusServlet",
"*.sink.prometheusServlet.path": "/metrics/prometheus",
"applications.sink.prometheusServlet.path": "/metrics/applications/prometheus",
"driver.source.jvm.class": "org.apache.spark.metrics.source.JvmSource",
"executor.source.jvm.class": "org.apache.spark.metrics.source.JvmSource",
"master.sink.prometheusServlet.path": "/metrics/master/prometheus",
"master.source.jvm.class": "org.apache.spark.metrics.source.JvmSource",
"worker.source.jvm.class": "org.apache.spark.metrics.source.JvmSource"
}
},
{
"Classification": "capacity-scheduler",
"Properties": {
"yarn.scheduler.capacity.resource-calculator": "org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator "
}
},
{
"Classification": "yarn-site",
"Properties": {
"yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage": "99.0",
"yarn.nodemanager.pmem-check-enabled": "false",
"yarn.nodemanager.vmem-check-enabled": "false"
}
},
{
"Classification": "emrfs-site",
"Properties": {
"fs.s3.maxConnections": "1000"
}
},
{
"Classification": "hive-site",
"Properties": {
"hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
}
},
{
"Classification": "hdfs-site",
"Properties": {
"dfs.replication": "2"
}
},
{
"Classification": "presto-connector-hive",
"Properties": {
"hive.metastore.glue.datacatalog.enabled": "true",
"hive.parquet.use-column-names": "true"
}
},
{
"Classification": "spark-hive-site",
"Properties": {
"hive.metastore.client.factory.class": "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
}
},
{
"Classification": "spark-env",
"Configurations": [
{
"Classification": "export",
"Properties": {
"PYSPARK_PYTHON": "./data_platform_spark_jobs.pex"
}
}
],
"Properties": {}
},
{
"Classification": "hadoop-env",
"Configurations": [
{
"Classification": "export",
"Properties": {
"HADOOP_DATANODE_OPTS": "-javaagent:/etc/prometheus/jmx_prometheus_javaagent.jar=7001:/etc/hadoop/conf/hdfs_jmx_config_datanode.yaml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.port=50103",
"HADOOP_NAMENODE_OPTS": "-javaagent:/etc/prometheus/jmx_prometheus_javaagent.jar=7001:/etc/hadoop/conf/hdfs_jmx_config_namenode.yaml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.port=50103"
}
}
],
"Properties": {}
},
{
"Classification": "yarn-env",
"Configurations": [
{
"Classification": "export",
"Properties": {
"YARN_NODEMANAGER_OPTS": "-javaagent:/etc/prometheus/jmx_prometheus_javaagent.jar=7005:/etc/hadoop/conf/yarn_jmx_config_node_manager.yaml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.port=50111",
"YARN_RESOURCEMANAGER_OPTS": "-javaagent:/etc/prometheus/jmx_prometheus_javaagent.jar=7005:/etc/hadoop/conf/yarn_jmx_config_resource_manager.yaml -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.port=50111"
}
}
],
"Properties": {}
},
{
"Classification": "hudi-defaults",
"Properties": {
"hoodie.archive.async": "true",
"hoodie.bulkinsert.sort.mode": "GLOBAL_SORT",
"hoodie.clean.async": "true",
"hoodie.cleaner.commits.retained": "1",
"hoodie.cleaner.policy.failed.writes": "LAZY",
"hoodie.datasource.hive_sync.support_timestamp": "true",
"hoodie.datasource.meta.sync.glue.metadata_file_listing": "true",
"hoodie.enable.data.skipping": "true",
"hoodie.filesystem.view.remote.retry.enable": "true",
"hoodie.keep.max.commits": "15",
"hoodie.keep.min.commits": "10",
"hoodie.metadata.index.bloom.filter.enable": "true",
"hoodie.metadata.index.column.stats.enable": "true",
"hoodie.metrics.on": "true",
"hoodie.metrics.reporter.type": "PROMETHEUS",
"hoodie.parquet.compression.codec": "snappy",
"hoodie.parquet.max.file.size": "536870912",
"hoodie.parquet.small.file.limit": "429496729",
"hoodie.write.concurrency.early.conflict.detection.enable": "true",
"hoodie.write.concurrency.mode": "optimistic_concurrency_control",
"hoodie.write.lock.dynamodb.billing_mode": "PAY_PER_REQUEST",
"hoodie.write.lock.dynamodb.endpoint_url": "dynamodb.eu-west-1.amazonaws.com",
"hoodie.write.lock.dynamodb.region": "eu-west-1",
"hoodie.write.lock.dynamodb.table": "data-platform-hudi-locks",
"hoodie.write.lock.provider": "org.apache.hudi.aws.transaction.lock.DynamoDBBasedLockProvider"
}
}
]with some additional dataset specific config:
hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.precombine.field=version
hoodie.datasource.write.partitionpath.field=story_published_partition_date
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.MultiPartKeysValueExtractor
hoodie.datasource.write.hive_style_partitioning=true
hoodie.avro.schema.validate=true
hoodie.datasource.write.reconcile.schema=false
hoodie.table.name=${TABLE_NAME}
hoodie.datasource.hive_sync.enable=true
hoodie.datasource.hive_sync.database=articles
hoodie.datasource.hive_sync.table=${TABLE_NAME}
hoodie.datasource.hive_sync.partition_fields=story_published_partition_date
hoodie.write.lock.dynamodb.partition_key=${TABLE_NAME}
# as the record key is random, don't try to prune by ranges
hoodie.bloom.index.prune.by.ranges=false
hoodie.index.type=RECORD_INDEX
hoodie.metadata.enable=true
hoodie.metadata.record.index.enable=true
danny0405
Describe the problem you faced
see this slack thread, I was told to raise an issue. I don't have a lot of time to debug this as the upgrade isn't essential right now
After upgrading Hudi from 0.12.1 to 0.13.1 via an EMR upgrade I’m seeing a lot of these when using the spark writer:
I’ve enabled retries, but it seems to be slowing down various write tasks a lot as they retry/fallover to secondary methods. Why would this be happening? Between these, and seemingly slower bloom filter lookups, jobs are taking 2x longer or more.
I'm unsure if these correspond to these warnings on the driver logs:
I’m also seeing similar errors on writes:
I had to rollback the upgrade as it was causing writes to fail (in addition to the successes taking 2x time)
To Reproduce
Unknown
Expected behavior
The performance to not degrade after upgrading.
Environment Description
Hudi version : 0.13.1-amzn-1 (EMR 6.13.0)
Spark version : 3.4.1
Hive version : 3.1.3
Hadoop version : 3.3.3
Storage (HDFS/S3/GCS..) : S3
Running on Docker? (yes/no) : no
Additional context
Upgrading from EMR emr-6.9.0 to emr-6.13.0.
This affected both tables we ingest, write times increased 2x for each cluster when succeeding, and failed for large writes.
EMR config:
Additional Hudi config:
The hours number seems to have become nonsense above (this is from persistent spark logs on EMR)