You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "abhisheksahani91 (via GitHub)" <gi...@apache.org> on 2024/08/08 08:43:09 UTC
[I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
abhisheksahani91 opened a new issue, #11739:
URL: https://github.com/apache/hudi/issues/11739
**_Tips before filing an issue_**
- Have you gone through our [FAQs](https://hudi.apache.org/learn/faq/)?
- Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.
- If you have triaged this as a bug, then file an [issue](https://issues.apache.org/jira/projects/HUDI/issues) directly.
**Describe the problem you faced**
A clear and concise description of the problem.
**To Reproduce**
Steps to reproduce the behavior:
1. Disable Executing compaction as a part Glue deltastreamer job
2. Write a separate glue job for executing the compaction by reading the scheduled compactions
**Expected behavior**
For the compaction glue job we have provided 5 executors, we are expecting it to run parallel on multiple executors. but only one executor is running the compaction.
**Environment Description**
AWS GLUE 4.0
* Hudi version : 0.12.1
* Spark version : 3.3
* Hive version :
* Hadoop version :
* Storage (HDFS/S3/GCS..) : s3
* Running on Docker? (yes/no) : no
**Additional context**
Problem Statement:
We have a separate AWS GLUE job for executing scheduled compactions. Even if we have 5 executors for glue compaction job, the compaction is executing on only one executor
`import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.MappingSpec
import com.amazonaws.services.glue.errors.CallSite
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._
import org.apache.spark.sql.SparkSession
import org.apache.spark.api.java.JavaSparkContext
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer
import org.apache.hudi.utilities.deltastreamer.SchedulerConfGenerator
import org.apache.hudi.utilities.UtilHelpers
import org.apache.hudi.utilities.HoodieCompactor
object GlueApp {
def main(sysArgs: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder
.appName("YourAppName")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.shuffle.partitions", "200") // Increase shuffle partitions for parallelism
.config("spark.default.parallelism", "200") // Increase default parallelism
.config("spark.dynamicAllocation.enabled", "true")
.config("spark.shuffle.service.enabled","true")
.getOrCreate()
val glueContext: GlueContext = new GlueContext(spark.sparkContext)
// Parsing command line arguments
val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME", "BASE-PATH", "TABLE-NAME", "SCHEMA-FILE").toArray)
// Initialize Glue job
Job.init(args("JOB_NAME"), glueContext, args.asJava)
val config = new HoodieCompactor.Config()
config.basePath = "s3://datalake/hudi_data_lake_prod10/prod_datalake_latest_v3/"
config.tableName = "prod_datalake_latest_v3"
config.schemaFile = "s3://datalake/schema/latest-prod-schema_v1.avsc"
// Create a JavaSparkContext using the existing SparkSession's SparkContext
val javaSparkContext = new JavaSparkContext(spark.sparkContext)
// Create an instance of HoodieCompactor
val hoodieCompactor = new HoodieCompactor(javaSparkContext,config)
// Invoke the doCompact method directly, passing the JavaSparkContext
hoodieCompactor.compact(1);
// Commit Glue job
Job.commit()
}
}`
[
<img width="1277" alt="Screenshot 2024-08-08 at 2 33 52 AM" src="https://github.com/user-attachments/assets/6fda0574-1383-41de-977c-932e8aafd62a">
<img width="1274" alt="Screenshot 2024-08-08 at 2 35 37 AM" src="https://github.com/user-attachments/assets/9e03e5c7-dbf1-46dd-b698-54a1801696b1">
<img width="1272" alt="Screenshot 2024-08-08 at 2 48 44 AM" src="https://github.com/user-attachments/assets/3c189cc2-b420-4bf1-ac06-db078d428100">
](url)
**Stacktrace**
```Add the stacktrace of the error.```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go commented on issue #11739:
URL: https://github.com/apache/hudi/issues/11739#issuecomment-2275818546
@abhisheksahani91 This stage is just preparing the compaction metadata by reading commit metadata from timeline. So it may be running in one task only.
Do you see single task in other stages also?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go closed issue #11739: [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE
URL: https://github.com/apache/hudi/issues/11739
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
Posted by "abhisheksahani91 (via GitHub)" <gi...@apache.org>.
abhisheksahani91 commented on issue #11739:
URL: https://github.com/apache/hudi/issues/11739#issuecomment-2276973660
@ad1happy2go
Attaching screenshot for stages ui
<img width="1277" alt="Screenshot 2024-08-09 at 6 40 34 AM" src="https://github.com/user-attachments/assets/80274e89-62a2-4983-9c9c-5a2f3859da13">
Please can you guide me on how I can debug to find out why stage 0 which is preparing compaction metadata is taking so much time?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
Posted by "ad1happy2go (via GitHub)" <gi...@apache.org>.
ad1happy2go commented on issue #11739:
URL: https://github.com/apache/hudi/issues/11739#issuecomment-2288741672
Discussed with @abhisheksahani91 on call. The issue was there was only one file group for which all the log files were there. The size of log files were also big so it created on one task and took so much time.
Confirmed with him, we can close this as it's not an issue. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [SUPPORT] How to achieve parallelism for Offline compaction job utilising utility HoodieCompactor.java running on aws GLUE [hudi]
Posted by "abhisheksahani91 (via GitHub)" <gi...@apache.org>.
abhisheksahani91 commented on issue #11739:
URL: https://github.com/apache/hudi/issues/11739#issuecomment-2277046725
@ad1happy2go
From the source code below function in class HoodieCompactor provides the responsibility for compaction
/**
* Execute compaction operations and report back status.
*/
public HoodieData<WriteStatus> compact(
HoodieEngineContext context, HoodieCompactionPlan compactionPlan,
HoodieTable table, HoodieWriteConfig config, String compactionInstantTime,
HoodieCompactionHandler compactionHandler) {
if (compactionPlan == null || (compactionPlan.getOperations() == null)
|| (compactionPlan.getOperations().isEmpty())) {
return context.emptyHoodieData();
}
HoodieActiveTimeline timeline = table.getActiveTimeline();
HoodieInstant instant = HoodieTimeline.getCompactionRequestedInstant(compactionInstantTime);
// Mark instant as compaction inflight
timeline.transitionCompactionRequestedToInflight(instant);
table.getMetaClient().reloadActiveTimeline();
HoodieTableMetaClient metaClient = table.getMetaClient();
TableSchemaResolver schemaResolver = new TableSchemaResolver(metaClient);
// Here we firstly use the table schema as the reader schema to read
// log file.That is because in the case of MergeInto, the config.getSchema may not
// the same with the table schema.
try {
if (StringUtils.isNullOrEmpty(config.getInternalSchema())) {
Schema readerSchema = schemaResolver.getTableAvroSchema(false);
config.setSchema(readerSchema.toString());
}
} catch (Exception e) {
// If there is no commit in the table, just ignore the exception.
}
// Compacting is very similar to applying updates to existing file
List<CompactionOperation> operations = compactionPlan.getOperations().stream()
.map(CompactionOperation::convertFromAvroRecordInstance).collect(toList());
LOG.info("Compactor compacting " + operations + " files");
context.setJobStatus(this.getClass().getSimpleName(), "Compacting file slices: " + config.getTableName());
TaskContextSupplier taskContextSupplier = table.getTaskContextSupplier();
return context.parallelize(operations).map(operation -> compact(
compactionHandler, metaClient, config, operation, compactionInstantTime, taskContextSupplier))
.flatMap(List::iterator);
}
from the logs, we can see that the following files are selected for compaction
24/08/07 20:40:21 INFO HoodieCompactor: Compactor compacting [CompactionOperation{baseInstantTime='20240807183515584', dataFileCommitTime=Option{val=20240807183515584}, deltaFileNames=[.1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.1_1-376-2470, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.2_1-435-2837, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.3_1-493-3203, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.4_1-551-3569, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.5_1-609-3935, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.6_1-667-4301, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.7_1-725-4667, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.8_1-783-5033, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.9_1-841-5399, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.10_1-912-5807, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_2024080
7183515584.log.11_1-970-6173, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.12_1-1030-6542, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.13_1-1088-6908, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.14_1-1146-7274, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.15_1-1204-7640, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.16_1-1275-8048, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.17_1-1333-8414, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.18_1-1391-8785, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.19_1-1449-9156, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.20_1-1507-9527, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.21_1-1565-9898, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.22_1-1623-10269, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.23_1-1668-10567, .1668c2db-6f1d-4b92-9ba9-d251
15fe3b74-0_20240807183515584.log.24_1-1726-10938, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.25_1-1784-11309, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.26_1-1842-11680, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.27_1-1900-12051, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.28_1-1971-12464, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.29_1-2029-12835, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.30_1-2100-13248, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.31_1-2171-13661, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.32_1-2228-14031, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.33_1-2285-14401, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.34_1-2343-14772, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.35_1-2401-15143, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.36_1-2459-15514,
.1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.37_1-2517-15885, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.38_1-2575-16256, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.39_1-2633-16627, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.40_1-2691-16998, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.41_1-2749-17369, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.42_1-2820-17782, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.43_1-2878-18153, .1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_20240807183515584.log.44_1-2936-18524], dataFileName=Option{val=1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0_0-0-0_20240807183515584.parquet}, id='HoodieFileGroupId{partitionPath='', fileId='1668c2db-6f1d-4b92-9ba9-d25115fe3b74-0'}', metrics={TOTAL_LOG_FILES=44.0, TOTAL_IO_READ_MB=4754.0, TOTAL_LOG_FILES_SIZE=4.875567816E9, TOTAL_IO_WRITE_MB=104.0, TOTAL_IO_MB=4858.0}, bootstrapFilePath=Optional.empty}] fi
les
Now if we focus on the code line
context.parallelize(operations).map(operation -> compact(
compactionHandler, metaClient, config, operation, compactionInstantTime, taskContextSupplier))
.flatMap(List::iterator);
context.parallelize(operations) is converting the list of files to HoodieJavaRDD
@Override
public <T> HoodieData<T> parallelize(List<T> data, int parallelism) {
return HoodieJavaRDD.of(javaSparkContext.parallelize(data, parallelism));
}
My doubt is if Hoodie is parallelizing the files selected for compaction then why the whole stage 0 is executed over a single executor/worker?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org