yihua commented on code in PR #9008:
URL: https://github.com/apache/hudi/pull/9008#discussion_r1235598864
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java:
##########
@@ -616,9 +619,9 @@ public static class DeltaSyncService extends
HoodieIngestionService {
private transient SparkSession sparkSession;
/**
- * Spark context.
+ * Spark context Wrapper.
*/
- private transient JavaSparkContext jssc;
+ private final transient HoodieSparkEngineContext sparkContext;
Review Comment:
similar here for renaming
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java:
##########
@@ -208,9 +208,9 @@ public class DeltaSync implements Serializable, Closeable {
private transient FileSystem fs;
/**
- * Spark context.
+ * Spark context Wrapper.
*/
- private transient JavaSparkContext jssc;
+ private final transient HoodieSparkEngineContext sparkContext;
Review Comment:
rename this to `hoodieSparkContext` or `hoodieSparkEngineContext`?
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/common/HoodieSparkEngineContext.java:
##########
@@ -211,4 +218,20 @@ public List<Integer>
removeCachedDataIds(HoodieDataCacheKey cacheKey) {
return removed == null ? Collections.emptyList() : removed;
}
}
+
+ public SparkConf getConf() {
+ return javaSparkContext.getConf();
+ }
+
+ public SparkConf getSparkConf() {
+ return javaSparkContext.sc().getConf();
+ }
Review Comment:
What's the difference between these two?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java:
##########
@@ -706,15 +709,15 @@ public DeltaSyncService(Config cfg, JavaSparkContext
jssc, FileSystem fs, Config
LOG.info(toSortedTruncatedString(props));
this.schemaProvider = UtilHelpers.wrapSchemaProviderWithPostProcessor(
- UtilHelpers.createSchemaProvider(cfg.schemaProviderClassName, props,
jssc), props, jssc, cfg.transformerClassNames);
+ UtilHelpers.createSchemaProvider(cfg.schemaProviderClassName, props,
sparkContext.jsc()),
+ props, sparkContext.jsc(), cfg.transformerClassNames);
- deltaSync = new DeltaSync(cfg, sparkSession, schemaProvider, props,
jssc, fs, conf,
- this::onInitializingWriteClient);
+ deltaSync = new DeltaSync(cfg, sparkSession, schemaProvider, props,
sparkContext, fs, conf, this::onInitializingWriteClient);
}
- public DeltaSyncService(HoodieDeltaStreamer.Config cfg, JavaSparkContext
jssc, FileSystem fs, Configuration conf)
+ public DeltaSyncService(HoodieDeltaStreamer.Config cfg,
HoodieSparkEngineContext sparkContext, FileSystem fs, Configuration conf)
Review Comment:
same here for all `HoodieSparkEngineContext` definition.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
