You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2020/08/04 05:32:56 UTC
[GitHub] [hudi] vinothchandar commented on a change in pull request #1810: [HUDI-875] Abstract hudi-sync-common, and support hudi-hive-sync
vinothchandar commented on a change in pull request #1810:
URL: https://github.com/apache/hudi/pull/1810#discussion_r464805558
##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
##########
@@ -267,9 +267,16 @@ public Operation convert(String value) throws ParameterException {
description = "Should duplicate records from source be dropped/filtered out before insert/bulk-insert")
public Boolean filterDupes = false;
+ //will abandon in the future version, recommended use --enable-sync
Review comment:
can we print a warning around this, so the user knows?
here's my take. we can change the code so that --enable-sync and --sync-tool-class-list are the main drivers out of which we derive a `Set<String>` denoting all the sync tool classes. if --enable-hive-sync is specified, then we simply add the hive sync tool class to this set.. rest of the code just syncs to all sync tools in this set.
this way, `--enable-hive-sync` will be just isolated to the initial command line parsing code. We can apply the same method to datasource as well, if you don't see issues @lw309637554 @leesf wdyt?
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
##########
@@ -261,6 +268,43 @@ private[hudi] object HoodieSparkSqlWriter {
hiveSyncConfig
}
+ private def metaSync(parameters: Map[String, String],
+ basePath: Path,
+ hadoopConf: Configuration): Boolean = {
+ val hiveSyncEnabled = parameters.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var metaSyncEnabled = parameters.get(HUDI_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var syncClientToolClass = parameters.get(SYNC_CLIENT_TOOL_CLASS).get
+ // for backward compatibility
+ if (hiveSyncEnabled) {
+ metaSyncEnabled = true
+ syncClientToolClass = DEFAULT_SYNC_CLIENT_TOOL_CLASS
+ }
+ var metaSyncSuccess = true
+ if (metaSyncEnabled) {
+ val impls = syncClientToolClass.split(",")
+ impls.foreach(impl => {
+ val syncSuccess = impl.trim match {
+ case DEFAULT_SYNC_CLIENT_TOOL_CLASS => {
+ log.info("Syncing to Hive Metastore (URL: " + parameters(HIVE_URL_OPT_KEY) + ")")
+ val fs = FSUtils.getFs(basePath.toString, hadoopConf)
+ syncHive(basePath, fs, parameters)
+ }
+ case _ => {
+ val fs = FSUtils.getFs(basePath.toString, hadoopConf)
Review comment:
can this line be shared . the `fs` initialization?
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
##########
@@ -261,6 +268,43 @@ private[hudi] object HoodieSparkSqlWriter {
hiveSyncConfig
}
+ private def metaSync(parameters: Map[String, String],
+ basePath: Path,
+ hadoopConf: Configuration): Boolean = {
+ val hiveSyncEnabled = parameters.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var metaSyncEnabled = parameters.get(HUDI_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var syncClientToolClass = parameters.get(SYNC_CLIENT_TOOL_CLASS).get
+ // for backward compatibility
+ if (hiveSyncEnabled) {
+ metaSyncEnabled = true
+ syncClientToolClass = DEFAULT_SYNC_CLIENT_TOOL_CLASS
+ }
+ var metaSyncSuccess = true
+ if (metaSyncEnabled) {
+ val impls = syncClientToolClass.split(",")
+ impls.foreach(impl => {
+ val syncSuccess = impl.trim match {
+ case DEFAULT_SYNC_CLIENT_TOOL_CLASS => {
+ log.info("Syncing to Hive Metastore (URL: " + parameters(HIVE_URL_OPT_KEY) + ")")
Review comment:
can we explicitly match for Hive instead of default? we may change the default for e.g and it would be an issue.
##########
File path: packaging/hudi-hive-sync-bundle/pom.xml
##########
@@ -66,7 +66,8 @@
<includes>
<include>org.apache.hudi:hudi-common</include>
<include>org.apache.hudi:hudi-hadoop-mr</include>
- <include>org.apache.hudi:hudi-hive-sync</include>
+ <include>org.apache.hudi:hudi-sync-common</include>
+ <include>org.apache.hudi:hudi-hive-sync</include>
Review comment:
nit:indent
##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
##########
@@ -442,7 +449,8 @@ private void refreshTimeline() throws IOException {
long overallTimeMs = overallTimerContext != null ? overallTimerContext.stop() : 0;
// Send DeltaStreamer Metrics
- metrics.updateDeltaStreamerMetrics(overallTimeMs, hiveSyncTimeMs);
+ metrics.updateDeltaStreamerMetrics(overallTimeMs, hiveSyncTimeMs, true);
+ metrics.updateDeltaStreamerMetrics(overallTimeMs, metaSyncTimeMs, false);
Review comment:
is there a way to do this by iterating over the configured sync tool classes? i.e only do it when sync is configured?
##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerMetrics.java
##########
@@ -67,10 +77,15 @@ String getMetricsName(String action, String metric) {
return config == null ? null : String.format("%s.%s.%s", tableName, action, metric);
}
- public void updateDeltaStreamerMetrics(long durationInNs, long hiveSyncNs) {
+ public void updateDeltaStreamerMetrics(long durationInNs, long syncNs, boolean hiveSync) {
if (config.isMetricsOn()) {
Metrics.registerGauge(getMetricsName("deltastreamer", "duration"), getDurationInMs(durationInNs));
- Metrics.registerGauge(getMetricsName("deltastreamer", "hiveSyncDuration"), getDurationInMs(hiveSyncNs));
+ if (hiveSync) {
+ Metrics.registerGauge(getMetricsName("deltastreamer", "hiveSyncDuration"), getDurationInMs(syncNs));
+ } else {
+ Metrics.registerGauge(getMetricsName("deltastreamer", "metaSyncDuration"), getDurationInMs(syncNs));
Review comment:
what if both hive and meta sync are off? we would still emit metrics for meta?
##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
##########
@@ -237,6 +237,9 @@ public Operation convert(String value) throws ParameterException {
@Parameter(names = {"--enable-hive-sync"}, description = "Enable syncing to hive")
public Boolean enableHiveSync = false;
+ @Parameter(names = {"--hoodie-sync-client-tool-class"}, description = "Meta sync client tool, using comma to separate multi tools")
Review comment:
I think we can name this little shorter. `--sync-tool-class-list` ?
##########
File path: hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
##########
@@ -255,6 +262,43 @@ private[hudi] object HoodieSparkSqlWriter {
hiveSyncConfig
}
+ private def metaSync(parameters: Map[String, String],
+ basePath: Path,
+ hadoopConf: Configuration): Boolean = {
+ val hiveSyncEnabled = parameters.get(HIVE_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var metaSyncEnabled = parameters.get(HUDI_SYNC_ENABLED_OPT_KEY).exists(r => r.toBoolean)
+ var syncClientToolClass = parameters.get(SYNC_CLIENT_TOOL_CLASS).get
+ // for backward compatibility
+ if (hiveSyncEnabled) {
+ metaSyncEnabled = true
+ syncClientToolClass = DEFAULT_SYNC_CLIENT_TOOL_CLASS
Review comment:
would users sync to both? down the line it may make sense to provide support for syncing to multiple things.
but even here, if we just append the HiveSync class when `hiveSyncEnabled=true`, we can support syncing to both Hive and dla?
##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerMetrics.java
##########
@@ -67,10 +77,15 @@ String getMetricsName(String action, String metric) {
return config == null ? null : String.format("%s.%s.%s", tableName, action, metric);
}
- public void updateDeltaStreamerMetrics(long durationInNs, long hiveSyncNs) {
+ public void updateDeltaStreamerMetrics(long durationInNs, long syncNs, boolean hiveSync) {
if (config.isMetricsOn()) {
Metrics.registerGauge(getMetricsName("deltastreamer", "duration"), getDurationInMs(durationInNs));
- Metrics.registerGauge(getMetricsName("deltastreamer", "hiveSyncDuration"), getDurationInMs(hiveSyncNs));
+ if (hiveSync) {
+ Metrics.registerGauge(getMetricsName("deltastreamer", "hiveSyncDuration"), getDurationInMs(syncNs));
+ } else {
+ Metrics.registerGauge(getMetricsName("deltastreamer", "metaSyncDuration"), getDurationInMs(syncNs));
Review comment:
should we derive the metric name from the sync tool class. i.e instead of `metaSyncDuration`, we do `dlaSyncDuration`? that seems more usable and understandable
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org