Skip to content

[HUDI-1089] Refactor hudi-client to support multi-engine #1827

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
vinothchandar merged 6 commits into from
Oct 1, 2020
Merged

[HUDI-1089] Refactor hudi-client to support multi-engine #1827

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b67ee14
[HUDI-1089] Refactor hudi-client to support multi-engine
wangxianghu Sep 22, 2020
cd5d75e
Code Review Comments
vinothchandar Sep 28, 2020
6a79819
More code review changes
vinothchandar Sep 30, 2020
7e33f73
Dropping HoodieIndex#fetchRecordLocation API
vinothchandar Oct 1, 2020
9d7a51d
Renaming some Abstract* classes as Hoodie* to improve readability
vinothchandar Oct 1, 2020
137da2b
fix Unable to find driver bind address from spark config
wangxianghu Oct 1, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
8 changes: 4 additions & 4 deletions .travis.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,10 @@ jdk:
- openjdk8
jobs:
include:
- name: "Unit tests except hudi-client"
env: MODE=unit MODULES='!hudi-client' HUDI_QUIETER_LOGGING=1
- name: "Unit tests for hudi-client"
env: MODE=unit MODULES=hudi-client HUDI_QUIETER_LOGGING=1
- name: "Unit tests except hudi-spark-client"
env: MODE=unit MODULES='!hudi-client/hudi-spark-client' HUDI_QUIETER_LOGGING=1
- name: "Unit tests for hudi-spark-client"
env: MODE=unit MODULES=hudi-client/hudi-spark-client HUDI_QUIETER_LOGGING=1
- name: "Functional tests"
env: MODE=functional HUDI_QUIETER_LOGGING=1
- name: "Integration tests"
Expand Down
9 changes: 8 additions & 1 deletion hudi-cli/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,14 @@
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client</artifactId>
<artifactId>hudi-client-common</artifactId>
<version>${project.version}</version>
<scope>test</scope>
<type>test-jar</type>
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-client</artifactId>

@yanghua yanghua Sep 24, 2020

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.

Actually, I am thinking hudi-spark-client and hudi-client-spark, which is better? Given there is a module named hudi-client-common. wdyt? @vinothchandar

@vinothchandar vinothchandar Sep 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

cc @wangxianghu probably hudi-client-spark is easier on the eyes?

@wangxianghu wangxianghu Sep 24, 2020

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.

cc @wangxianghu probably hudi-client-spark is easier on the eyes?

I am ok with both of them :) let's rename it to hudi-client-spark

@yanghua yanghua Oct 1, 2020 •
edited
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.

So, will we keep hudi-spark-client? We have had a hudi-spark module. IMHO, this naming may not seem so clear.

@wangxianghu wangxianghu Oct 2, 2020

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.

So, will we keep hudi-spark-client? We have had a hudi-spark module. IMHO, this naming may not seem so clear.

@vinothchandar @yanghua I have filed a pr to rename hudi-spark-client to hudi-client-spark : #2139
please take a look when free

<version>${project.version}</version>
<scope>test</scope>
<type>test-jar</type>
Expand Down
7 changes: 4 additions & 3 deletions hudi-cli/src/main/java/org/apache/hudi/cli/commands/SavepointsCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@
import org.apache.hudi.cli.HoodieTableHeaderFields;
import org.apache.hudi.cli.utils.InputStreamConsumer;
import org.apache.hudi.cli.utils.SparkUtil;
import org.apache.hudi.client.HoodieWriteClient;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
Expand Down Expand Up @@ -162,10 +163,10 @@ public String deleteSavepoint(@CliOption(key = {"commit"}, help = "Delete a save
return String.format("Savepoint \"%s\" deleted.", instantTime);
}

private static HoodieWriteClient createHoodieClient(JavaSparkContext jsc, String basePath) throws Exception {
private static SparkRDDWriteClient createHoodieClient(JavaSparkContext jsc, String basePath) throws Exception {
HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.BLOOM).build()).build();
return new HoodieWriteClient(jsc, config, false);
return new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), config, false);
}

}
24 changes: 13 additions & 11 deletions hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,21 +22,22 @@
import org.apache.hudi.DataSourceWriteOptions;
import org.apache.hudi.cli.DedupeSparkJob;
import org.apache.hudi.cli.utils.SparkUtil;
import org.apache.hudi.client.HoodieWriteClient;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.client.utils.ClientUtils;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieBootstrapConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieSavepointException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.upgrade.UpgradeDowngrade;
import org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy;
import org.apache.hudi.table.upgrade.SparkUpgradeDowngrade;
import org.apache.hudi.utilities.HDFSParquetImporter;
import org.apache.hudi.utilities.HDFSParquetImporter.Config;
import org.apache.hudi.utilities.HoodieCleaner;
Expand Down Expand Up @@ -343,7 +344,7 @@ private static int doBootstrap(JavaSparkContext jsc, String tableName, String ta
}

private static int rollback(JavaSparkContext jsc, String instantTime, String basePath) throws Exception {
HoodieWriteClient client = createHoodieClient(jsc, basePath);
SparkRDDWriteClient client = createHoodieClient(jsc, basePath);
if (client.rollback(instantTime)) {
LOG.info(String.format("The commit \"%s\" rolled back.", instantTime));
return 0;
Expand All @@ -355,7 +356,7 @@ private static int rollback(JavaSparkContext jsc, String instantTime, String bas

private static int createSavepoint(JavaSparkContext jsc, String commitTime, String user,
String comments, String basePath) throws Exception {
HoodieWriteClient client = createHoodieClient(jsc, basePath);
SparkRDDWriteClient client = createHoodieClient(jsc, basePath);
try {
client.savepoint(commitTime, user, comments);
LOG.info(String.format("The commit \"%s\" has been savepointed.", commitTime));
Expand All @@ -367,7 +368,7 @@ private static int createSavepoint(JavaSparkContext jsc, String commitTime, Stri
}

private static int rollbackToSavepoint(JavaSparkContext jsc, String savepointTime, String basePath) throws Exception {
HoodieWriteClient client = createHoodieClient(jsc, basePath);
SparkRDDWriteClient client = createHoodieClient(jsc, basePath);
try {
client.restoreToSavepoint(savepointTime);
LOG.info(String.format("The commit \"%s\" rolled back.", savepointTime));
Expand All @@ -379,7 +380,7 @@ private static int rollbackToSavepoint(JavaSparkContext jsc, String savepointTim
}

private static int deleteSavepoint(JavaSparkContext jsc, String savepointTime, String basePath) throws Exception {
HoodieWriteClient client = createHoodieClient(jsc, basePath);
SparkRDDWriteClient client = createHoodieClient(jsc, basePath);
try {
client.deleteSavepoint(savepointTime);
LOG.info(String.format("Savepoint \"%s\" deleted.", savepointTime));
Expand All @@ -401,9 +402,10 @@ private static int deleteSavepoint(JavaSparkContext jsc, String savepointTime, S
*/
protected static int upgradeOrDowngradeTable(JavaSparkContext jsc, String basePath, String toVersion) {
HoodieWriteConfig config = getWriteConfig(basePath);
HoodieTableMetaClient metaClient = ClientUtils.createMetaClient(jsc.hadoopConfiguration(), config, false);
HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), config.getBasePath(), false,
config.getConsistencyGuardConfig(), Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())));
try {
UpgradeDowngrade.run(metaClient, HoodieTableVersion.valueOf(toVersion), config, jsc, null);
new SparkUpgradeDowngrade(metaClient, config, new HoodieSparkEngineContext(jsc)).run(metaClient, HoodieTableVersion.valueOf(toVersion), config, new HoodieSparkEngineContext(jsc), null);
LOG.info(String.format("Table at \"%s\" upgraded / downgraded to version \"%s\".", basePath, toVersion));
return 0;
} catch (Exception e) {
Expand All @@ -412,9 +414,9 @@ protected static int upgradeOrDowngradeTable(JavaSparkContext jsc, String basePa
}
}

private static HoodieWriteClient createHoodieClient(JavaSparkContext jsc, String basePath) throws Exception {
private static SparkRDDWriteClient createHoodieClient(JavaSparkContext jsc, String basePath) throws Exception {
HoodieWriteConfig config = getWriteConfig(basePath);
return new HoodieWriteClient(jsc, config);
return new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), config);
}

private static HoodieWriteConfig getWriteConfig(String basePath) {
Expand Down
4 changes: 2 additions & 2 deletions hudi-cli/src/main/java/org/apache/hudi/cli/utils/SparkUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import org.apache.hudi.cli.HoodieCliSparkConfig;
import org.apache.hudi.cli.commands.SparkEnvCommand;
import org.apache.hudi.cli.commands.SparkMain;
import org.apache.hudi.client.HoodieWriteClient;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
Expand Down Expand Up @@ -91,7 +91,7 @@ public static JavaSparkContext initJavaSparkConf(String name, Option<String> mas
sparkConf.set(HoodieCliSparkConfig.CLI_MAPRED_OUTPUT_COMPRESSION_CODEC, "org.apache.hadoop.io.compress.GzipCodec");
sparkConf.set(HoodieCliSparkConfig.CLI_MAPRED_OUTPUT_COMPRESSION_TYPE, "BLOCK");

HoodieWriteClient.registerClasses(sparkConf);
SparkRDDWriteClient.registerClasses(sparkConf);
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
jsc.hadoopConfiguration().setBoolean(HoodieCliSparkConfig.CLI_PARQUET_ENABLE_SUMMARY_METADATA, false);
FSUtils.prepareHadoopConf(jsc.hadoopConfiguration());
Expand Down
6 changes: 4 additions & 2 deletions hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestArchivedCommitsCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTimelineArchiveLog;

import org.junit.jupiter.api.AfterEach;
Expand Down Expand Up @@ -92,8 +93,9 @@ public void init() throws Exception {
metaClient.getActiveTimeline().reload().getAllCommitsTimeline().filterCompletedInstants();

// archive
HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, hadoopConf);
archiveLog.archiveIfRequired(jsc);
HoodieSparkTable table = HoodieSparkTable.create(cfg, context, metaClient);

@vinothchandar vinothchandar Sep 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

can we replace more of the code to direclty just use HoodieTable instead. Need to examine cases that need an explicit HoodieSparkTable

@wangxianghu wangxianghu Sep 28, 2020

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.

can we replace more of the code to direclty just use HoodieTable instead. Need to examine cases that need an explicit HoodieSparkTable

yes, make sense

HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
archiveLog.archiveIfRequired(context);
}

@AfterEach
Expand Down
6 changes: 4 additions & 2 deletions hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTimelineArchiveLog;

import org.apache.hadoop.fs.FileSystem;
Expand Down Expand Up @@ -179,8 +180,9 @@ public void testShowArchivedCommits() throws Exception {

// archive
metaClient = HoodieTableMetaClient.reload(HoodieCLI.getTableMetaClient());
HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, jsc.hadoopConfiguration());
archiveLog.archiveIfRequired(jsc);
HoodieSparkTable table = HoodieSparkTable.create(cfg, context, metaClient);
HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
archiveLog.archiveIfRequired(context);

CommandResult cr = getShell().executeCommand(String.format("commits showarchived --startTs %s --endTs %s", "100", "104"));
assertTrue(cr.isSuccess());
Expand Down
4 changes: 2 additions & 2 deletions hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRollbacksCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import org.apache.hudi.cli.HoodieTableHeaderFields;
import org.apache.hudi.cli.TableHeader;
import org.apache.hudi.cli.testutils.AbstractShellIntegrationTest;
import org.apache.hudi.client.HoodieWriteClient;
import org.apache.hudi.client.AbstractHoodieWriteClient;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
Expand Down Expand Up @@ -90,7 +90,7 @@ public void init() throws Exception {
HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(tablePath)
.withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build()).build();

try (HoodieWriteClient client = getHoodieWriteClient(config)) {
try (AbstractHoodieWriteClient client = getHoodieWriteClient(config)) {

@vinothchandar vinothchandar Sep 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

just like this, we should try to use the abstract class as much as we can

// Rollback inflight commit3 and commit2
client.rollback("102");
client.rollback("101");
Expand Down
Loading