Skip to content

[HUDI-1269] Make whether the failure of connect hive affects hudi ingest process configurable #2443

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nsivabalan merged 7 commits into from
Feb 25, 2021
Merged

[HUDI-1269] Make whether the failure of connect hive affects hudi ingest process configurable #2443

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0d98db7
HUDI-1269
liujinhui1994 Jan 14, 2021
227c817
Merge branch 'master' into HUDI-1269
liujinhui1994 Jan 25, 2021
86faca0
Fix NullPointerException And Verified in CDH 6.3.0 environment
liujinhui1994 Feb 20, 2021
56f3f68
Merge branch 'master' into HUDI-1269
liujinhui1994 Feb 20, 2021
672445f
Adding tests
nsivabalan Feb 22, 2021
8aa5ce7
Fixing tests
nsivabalan Feb 23, 2021
7baf5de
Addressing feedback
nsivabalan Feb 25, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/DataSourceUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -293,6 +293,8 @@ public static HiveSyncConfig buildHiveSyncConfig(TypedProperties props, String b
DataSourceWriteOptions.DEFAULT_HIVE_USE_JDBC_OPT_VAL()));
hiveSyncConfig.autoCreateDatabase = Boolean.valueOf(props.getString(DataSourceWriteOptions.HIVE_AUTO_CREATE_DATABASE_OPT_KEY(),
DataSourceWriteOptions.DEFAULT_HIVE_AUTO_CREATE_DATABASE_OPT_KEY()));
hiveSyncConfig.ignoreConnectException = Boolean.valueOf(props.getString(DataSourceWriteOptions.HIVE_IGNORE_CONNECT_EXCEPTION_OPT_KEY(),

@yanghua yanghua Jan 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we extend a larger range from connect exception to sync exception?

@nsivabalan nsivabalan Feb 1, 2021 •
edited by yanghua
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yanghua : are you suggesting, we can make this config cover exceptions in syncHoodieTable() as well ?

@yanghua yanghua Feb 2, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yanghua : are you suggesting, we can make this config cover exceptions in syncHoodieTable() as well ?

Yes, do you have any concerns? In our scenario, we met all hudi spark jobs failed when the hive server is not available.

@nsivabalan nsivabalan Feb 4, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nope. was just clarifying :)

@nsivabalan nsivabalan Feb 21, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@liujinhui1994 : do you have any concerns to make this cover all exception w/ hive sync? I don't see you have addressed this feedback.

DataSourceWriteOptions.DEFAULT_HIVE_IGNORE_CONNECT_EXCEPTION_OPT_KEY()));
hiveSyncConfig.skipROSuffix = Boolean.valueOf(props.getString(DataSourceWriteOptions.HIVE_SKIP_RO_SUFFIX(),
DataSourceWriteOptions.DEFAULT_HIVE_SKIP_RO_SUFFIX_VAL()));
hiveSyncConfig.supportTimestamp = Boolean.valueOf(props.getString(DataSourceWriteOptions.HIVE_SUPPORT_TIMESTAMP(),
Expand Down
2 changes: 2 additions & 0 deletions ...spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,6 +307,7 @@ object DataSourceWriteOptions {
val HIVE_USE_PRE_APACHE_INPUT_FORMAT_OPT_KEY = "hoodie.datasource.hive_sync.use_pre_apache_input_format"
val HIVE_USE_JDBC_OPT_KEY = "hoodie.datasource.hive_sync.use_jdbc"
val HIVE_AUTO_CREATE_DATABASE_OPT_KEY = "hoodie.datasource.hive_sync.auto_create_database"
val HIVE_IGNORE_CONNECT_EXCEPTION_OPT_KEY = "hoodie.datasource.hive_sync.ignore_connect_exception"
val HIVE_SKIP_RO_SUFFIX = "hoodie.datasource.hive_sync.skip_ro_suffix"
val HIVE_SUPPORT_TIMESTAMP = "hoodie.datasource.hive_sync.support_timestamp"

Expand All @@ -325,6 +326,7 @@ object DataSourceWriteOptions {
val DEFAULT_USE_PRE_APACHE_INPUT_FORMAT_OPT_VAL = "false"
val DEFAULT_HIVE_USE_JDBC_OPT_VAL = "true"
val DEFAULT_HIVE_AUTO_CREATE_DATABASE_OPT_KEY = "true"
val DEFAULT_HIVE_IGNORE_CONNECT_EXCEPTION_OPT_KEY = "false"
val DEFAULT_HIVE_SKIP_RO_SUFFIX_VAL = "false"
val DEFAULT_HIVE_SUPPORT_TIMESTAMP = "false"

Expand Down
1 change: 1 addition & 0 deletions hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieSparkSqlWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,6 +373,7 @@ private[hudi] object HoodieSparkSqlWriter {
hiveSyncConfig.useJdbc = parameters(HIVE_USE_JDBC_OPT_KEY).toBoolean
hiveSyncConfig.useFileListingFromMetadata = parameters(HoodieMetadataConfig.METADATA_ENABLE_PROP).toBoolean
hiveSyncConfig.verifyMetadataFileListing = parameters(HoodieMetadataConfig.METADATA_VALIDATE_PROP).toBoolean
hiveSyncConfig.ignoreConnectException = parameters.get(HIVE_IGNORE_CONNECT_EXCEPTION_OPT_KEY).exists(r => r.toBoolean)
hiveSyncConfig.supportTimestamp = parameters.get(HIVE_SUPPORT_TIMESTAMP).exists(r => r.toBoolean)
hiveSyncConfig.autoCreateDatabase = parameters.get(HIVE_AUTO_CREATE_DATABASE_OPT_KEY).exists(r => r.toBoolean)
hiveSyncConfig.decodePartition = parameters.getOrElse(URL_ENCODE_PARTITIONING_OPT_KEY,
Expand Down
4 changes: 4 additions & 0 deletions hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ public class HiveSyncConfig implements Serializable {
@Parameter(names = {"--auto-create-database"}, description = "Auto create hive database")
public Boolean autoCreateDatabase = true;

@Parameter(names = {"--ignore-connect-exception"}, description = "Ignore connect hive exception")
public Boolean ignoreConnectException = false;

@Parameter(names = {"--skip-ro-suffix"}, description = "Skip the `_ro` suffix for Read optimized table, when registering")
public Boolean skipROSuffix = false;

Expand Down Expand Up @@ -130,6 +133,7 @@ public String toString() {
+ ", usePreApacheInputFormat=" + usePreApacheInputFormat
+ ", useJdbc=" + useJdbc
+ ", autoCreateDatabase=" + autoCreateDatabase
+ ", ignoreConnectException=" + ignoreConnectException
+ ", skipROSuffix=" + skipROSuffix
+ ", help=" + help
+ ", supportTimestamp=" + supportTimestamp
Expand Down
74 changes: 45 additions & 29 deletions hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,56 +57,72 @@ public class HiveSyncTool extends AbstractSyncTool {
public static final String SUFFIX_READ_OPTIMIZED_TABLE = "_ro";

private final HiveSyncConfig cfg;
private final HoodieHiveClient hoodieHiveClient;
private final String snapshotTableName;
private final Option<String> roTableTableName;
private HoodieHiveClient hoodieHiveClient = null;
private String snapshotTableName = null;
private Option<String> roTableTableName = null;

@yanghua yanghua Feb 20, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about roTableName?

@liujinhui1994 liujinhui1994 Feb 20, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it’s better to keep the original name

@nsivabalan nsivabalan Feb 21, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

guess it's a typo. Probably we can fix it while you are at it.


public HiveSyncTool(HiveSyncConfig cfg, HiveConf configuration, FileSystem fs) {
super(configuration.getAllProperties(), fs);
this.hoodieHiveClient = new HoodieHiveClient(cfg, configuration, fs);

try {
this.hoodieHiveClient = new HoodieHiveClient(cfg, configuration, fs);
Comment thread
nsivabalan marked this conversation as resolved.
} catch (RuntimeException e) {
if (cfg.ignoreConnectException) {
LOG.error("Got runtime exception when hive syncing", e);
Comment thread
nsivabalan marked this conversation as resolved.
Outdated
} else {
throw new HoodieHiveSyncException("Got runtime exception when hive syncing", e);
}
}

this.cfg = cfg;
// Set partitionFields to empty, when the NonPartitionedExtractor is used
if (NonPartitionedExtractor.class.getName().equals(cfg.partitionValueExtractorClass)) {
LOG.warn("Set partitionFields to empty, since the NonPartitionedExtractor is used");
cfg.partitionFields = new ArrayList<>();
}
switch (hoodieHiveClient.getTableType()) {
case COPY_ON_WRITE:
this.snapshotTableName = cfg.tableName;
this.roTableTableName = Option.empty();
break;
case MERGE_ON_READ:
this.snapshotTableName = cfg.tableName + SUFFIX_SNAPSHOT_TABLE;
this.roTableTableName = cfg.skipROSuffix ? Option.of(cfg.tableName) :
Option.of(cfg.tableName + SUFFIX_READ_OPTIMIZED_TABLE);
break;
default:
LOG.error("Unknown table type " + hoodieHiveClient.getTableType());
throw new InvalidTableException(hoodieHiveClient.getBasePath());
}
}

@Override
public void syncHoodieTable() {
try {
if (hoodieHiveClient != null) {
switch (hoodieHiveClient.getTableType()) {
case COPY_ON_WRITE:
syncHoodieTable(snapshotTableName, false);
this.snapshotTableName = cfg.tableName;
this.roTableTableName = Option.empty();
break;
case MERGE_ON_READ:
// sync a RO table for MOR
syncHoodieTable(roTableTableName.get(), false);
// sync a RT table for MOR
syncHoodieTable(snapshotTableName, true);
this.snapshotTableName = cfg.tableName + SUFFIX_SNAPSHOT_TABLE;
this.roTableTableName = cfg.skipROSuffix ? Option.of(cfg.tableName) :
Option.of(cfg.tableName + SUFFIX_READ_OPTIMIZED_TABLE);
break;
default:
LOG.error("Unknown table type " + hoodieHiveClient.getTableType());
throw new InvalidTableException(hoodieHiveClient.getBasePath());
}
}
}

@Override
public void syncHoodieTable() {
try {
if (hoodieHiveClient != null) {
switch (hoodieHiveClient.getTableType()) {
case COPY_ON_WRITE:
syncHoodieTable(snapshotTableName, false);
break;
case MERGE_ON_READ:
// sync a RO table for MOR
syncHoodieTable(roTableTableName.get(), false);
// sync a RT table for MOR
syncHoodieTable(snapshotTableName, true);
break;
default:
LOG.error("Unknown table type " + hoodieHiveClient.getTableType());
throw new InvalidTableException(hoodieHiveClient.getBasePath());
}
}
} catch (RuntimeException re) {
LOG.error("Got runtime exception when hive syncing", re);
} finally {
hoodieHiveClient.close();
if (hoodieHiveClient != null) {
hoodieHiveClient.close();
}
}
}

Expand Down