hostAddr = context.getProperty(EngineProperty.EMBEDDED_SERVER_HOST);
+ timelineServer = Option.of(new EmbeddedTimelineService(context, hostAddr.orElse(null),
config.getClientSpecifiedViewStorageConfig()));
try {
timelineServer.get().startServer();
@@ -122,6 +125,8 @@ public abstract class AbstractHoodieClient implements Serializable, AutoCloseabl
}
protected HoodieTableMetaClient createMetaClient(boolean loadActiveTimelineOnLoad) {
- return ClientUtils.createMetaClient(hadoopConf, config, loadActiveTimelineOnLoad);
+ return new HoodieTableMetaClient(hadoopConf, config.getBasePath(), loadActiveTimelineOnLoad,
+ config.getConsistencyGuardConfig(),
+ Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())));
}
}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
similarity index 61%
rename from hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
index 7970623ee..0f35e270e 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AbstractHoodieWriteClient.java
@@ -18,22 +18,27 @@
package org.apache.hudi.client;
+import com.codahale.metrics.Timer;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.callback.HoodieWriteCommitCallback;
+import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
+import org.apache.hudi.callback.util.HoodieCommitCallbackFactory;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieCompactionConfig;
@@ -45,109 +50,178 @@ import org.apache.hudi.exception.HoodieRollbackException;
import org.apache.hudi.exception.HoodieSavepointException;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.metrics.HoodieMetrics;
+import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.HoodieTimelineArchiveLog;
import org.apache.hudi.table.MarkerFiles;
-import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.action.HoodieWriteMetadata;
-import org.apache.hudi.table.action.compact.CompactHelpers;
import org.apache.hudi.table.action.savepoint.SavepointHelpers;
-
-import com.codahale.metrics.Timer;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
- * Hoodie Write Client helps you build tables on HDFS [insert()] and then perform efficient mutations on an HDFS
- * table [upsert()]
- *
- * Note that, at any given time, there can only be one Spark job performing these operations on a Hoodie table.
+ * Abstract Write Client providing functionality for performing commit, index updates and rollback
+ * Reused for regular write operations like upsert/insert/bulk-insert.. as well as bootstrap
+ *
+ * @param Sub type of HoodieRecordPayload
+ * @param Type of inputs
+ * @param Type of keys
+ * @param Type of outputs
*/
-public class HoodieWriteClient extends AbstractHoodieWriteClient {
+public abstract class AbstractHoodieWriteClient extends AbstractHoodieClient {
+ protected static final String LOOKUP_STR = "lookup";
private static final long serialVersionUID = 1L;
- private static final Logger LOG = LogManager.getLogger(HoodieWriteClient.class);
- private static final String LOOKUP_STR = "lookup";
- private final boolean rollbackPending;
- private final transient HoodieMetrics metrics;
- private transient Timer.Context compactionTimer;
- private transient AsyncCleanerService asyncCleanerService;
+ private static final Logger LOG = LogManager.getLogger(AbstractHoodieWriteClient.class);
+
+ protected final transient HoodieMetrics metrics;
+ private final transient HoodieIndex index;
+
+ protected transient Timer.Context writeTimer = null;
+ protected transient Timer.Context compactionTimer;
+
+ private transient WriteOperationType operationType;
+ private transient HoodieWriteCommitCallback commitCallback;
+ protected final boolean rollbackPending;
+ protected transient AsyncCleanerService asyncCleanerService;
/**
* Create a write client, without cleaning up failed/inflight commits.
*
- * @param jsc Java Spark Context
+ * @param context HoodieEngineContext
* @param clientConfig instance of HoodieWriteConfig
*/
- public HoodieWriteClient(JavaSparkContext jsc, HoodieWriteConfig clientConfig) {
- this(jsc, clientConfig, false);
+ public AbstractHoodieWriteClient(HoodieEngineContext context, HoodieWriteConfig clientConfig) {
+ this(context, clientConfig, false);
}
/**
* Create a write client, with new hudi index.
*
- * @param jsc Java Spark Context
+ * @param context HoodieEngineContext
* @param writeConfig instance of HoodieWriteConfig
* @param rollbackPending whether need to cleanup pending commits
*/
- public HoodieWriteClient(JavaSparkContext jsc, HoodieWriteConfig writeConfig, boolean rollbackPending) {
- this(jsc, writeConfig, rollbackPending, HoodieIndex.createIndex(writeConfig));
- }
-
- public HoodieWriteClient(JavaSparkContext jsc, HoodieWriteConfig writeConfig, boolean rollbackPending, HoodieIndex index) {
- this(jsc, writeConfig, rollbackPending, index, Option.empty());
+ public AbstractHoodieWriteClient(HoodieEngineContext context, HoodieWriteConfig writeConfig, boolean rollbackPending) {
+ this(context, writeConfig, rollbackPending, Option.empty());
}
/**
- * Create a write client, allows to specify all parameters.
+ * Create a write client, allows to specify all parameters.
*
- * @param jsc Java Spark Context
+ * @param context HoodieEngineContext
* @param writeConfig instance of HoodieWriteConfig
* @param rollbackPending whether need to cleanup pending commits
* @param timelineService Timeline Service that runs as part of write client.
*/
- public HoodieWriteClient(JavaSparkContext jsc, HoodieWriteConfig writeConfig, boolean rollbackPending,
- HoodieIndex index, Option timelineService) {
- super(jsc, index, writeConfig, timelineService);
+ public AbstractHoodieWriteClient(HoodieEngineContext context, HoodieWriteConfig writeConfig, boolean rollbackPending,
+ Option timelineService) {
+ super(context, writeConfig, timelineService);
this.metrics = new HoodieMetrics(config, config.getTableName());
this.rollbackPending = rollbackPending;
+ this.index = createIndex(writeConfig);
+ }
+
+ protected abstract HoodieIndex createIndex(HoodieWriteConfig writeConfig);
+
+ public void setOperationType(WriteOperationType operationType) {
+ this.operationType = operationType;
+ }
+
+ public WriteOperationType getOperationType() {
+ return this.operationType;
}
/**
- * Register hudi classes for Kryo serialization.
- *
- * @param conf instance of SparkConf
- * @return SparkConf
+ * Commit changes performed at the given instantTime marker.
*/
- public static SparkConf registerClasses(SparkConf conf) {
- conf.registerKryoClasses(new Class[]{HoodieWriteConfig.class, HoodieRecord.class, HoodieKey.class});
- return conf;
+ public boolean commit(String instantTime, O writeStatuses) {
+ return commit(instantTime, writeStatuses, Option.empty());
+ }
+
+ /**
+ *
+ * Commit changes performed at the given instantTime marker.
+ */
+ public boolean commit(String instantTime, O writeStatuses, Option> extraMetadata) {
+ HoodieTableMetaClient metaClient = createMetaClient(false);
+ String actionType = metaClient.getCommitActionType();
+ return commit(instantTime, writeStatuses, extraMetadata, actionType, Collections.emptyMap());
+ }
+
+ public abstract boolean commit(String instantTime, O writeStatuses, Option> extraMetadata,
+ String commitActionType, Map> partitionToReplacedFileIds);
+
+ public boolean commitStats(String instantTime, List stats, Option> extraMetadata,
+ String commitActionType) {
+ return commitStats(instantTime, stats, extraMetadata, commitActionType, Collections.emptyMap());
+ }
+
+ public boolean commitStats(String instantTime, List stats, Option> extraMetadata,
+ String commitActionType, Map> partitionToReplaceFileIds) {
+ LOG.info("Committing " + instantTime + " action " + commitActionType);
+ // Create a Hoodie table which encapsulated the commits and files visible
+ HoodieTable table = createTable(config, hadoopConf);
+
+ HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ HoodieCommitMetadata metadata = CommitUtils.buildMetadata(stats, partitionToReplaceFileIds, extraMetadata, operationType, config.getSchema(), commitActionType);
+ // Finalize write
+ finalizeWrite(table, instantTime, stats);
+
+ try {
+ activeTimeline.saveAsComplete(new HoodieInstant(true, commitActionType, instantTime),
+ Option.of(metadata.toJsonString().getBytes(StandardCharsets.UTF_8)));
+ postCommit(table, metadata, instantTime, extraMetadata);
+ emitCommitMetrics(instantTime, metadata, commitActionType);
+ LOG.info("Committed " + instantTime);
+ } catch (IOException e) {
+ throw new HoodieCommitException("Failed to complete commit " + config.getBasePath() + " at time " + instantTime,
+ e);
+ }
+
+ // callback if needed.
+ if (config.writeCommitCallbackOn()) {
+ if (null == commitCallback) {
+ commitCallback = HoodieCommitCallbackFactory.create(config);
+ }
+ commitCallback.call(new HoodieWriteCommitCallbackMessage(instantTime, config.getTableName(), config.getBasePath()));
+ }
+ return true;
+ }
+
+ protected abstract HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf);
+
+ void emitCommitMetrics(String instantTime, HoodieCommitMetadata metadata, String actionType) {
+ try {
+
+ if (writeTimer != null) {
+ long durationInMs = metrics.getDurationInMs(writeTimer.stop());
+ metrics.updateCommitMetrics(HoodieActiveTimeline.COMMIT_FORMATTER.parse(instantTime).getTime(), durationInMs,
+ metadata, actionType);
+ writeTimer = null;
+ }
+ } catch (ParseException e) {
+ throw new HoodieCommitException("Failed to complete commit " + config.getBasePath() + " at time " + instantTime
+ + "Instant time is not of valid format", e);
+ }
}
/**
* Filter out HoodieRecords that already exists in the output folder. This is useful in deduplication.
*
- * @param hoodieRecords Input RDD of Hoodie records.
- * @return A subset of hoodieRecords RDD, with existing records filtered out.
+ * @param hoodieRecords Input Hoodie records.
+ * @return A subset of hoodieRecords, with existing records filtered out.
*/
- public JavaRDD> filterExists(JavaRDD> hoodieRecords) {
- // Create a Hoodie table which encapsulated the commits and files visible
- HoodieTable table = HoodieTable.create(config, hadoopConf);
- Timer.Context indexTimer = metrics.getIndexCtx();
- JavaRDD> recordsWithLocation = getIndex().tagLocation(hoodieRecords, jsc, table);
- metrics.updateIndexMetrics(LOOKUP_STR, metrics.getDurationInMs(indexTimer == null ? 0L : indexTimer.stop()));
- return recordsWithLocation.filter(v1 -> !v1.isCurrentLocationKnown());
- }
+ public abstract I filterExists(I hoodieRecords);
/**
* Main API to run bootstrap to hudi.
@@ -156,8 +230,8 @@ public class HoodieWriteClient extends AbstractHo
if (rollbackPending) {
rollBackInflightBootstrap();
}
- HoodieTable table = getTableAndInitCtx(WriteOperationType.UPSERT, HoodieTimeline.METADATA_BOOTSTRAP_INSTANT_TS);
- table.bootstrap(jsc, extraMetadata);
+ HoodieTable table = getTableAndInitCtx(WriteOperationType.UPSERT, HoodieTimeline.METADATA_BOOTSTRAP_INSTANT_TS);
+ table.bootstrap(context, extraMetadata);
}
/**
@@ -165,14 +239,14 @@ public class HoodieWriteClient extends AbstractHo
*/
protected void rollBackInflightBootstrap() {
LOG.info("Rolling back pending bootstrap if present");
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
HoodieTimeline inflightTimeline = table.getMetaClient().getCommitsTimeline().filterPendingExcludingCompaction();
Option instant = Option.fromJavaOptional(
inflightTimeline.getReverseOrderedInstants().map(HoodieInstant::getTimestamp).findFirst());
if (instant.isPresent() && HoodieTimeline.compareTimestamps(instant.get(), HoodieTimeline.LESSER_THAN_OR_EQUALS,
HoodieTimeline.FULL_BOOTSTRAP_INSTANT_TS)) {
LOG.info("Found pending bootstrap instants. Rolling them back");
- table.rollbackBootstrap(jsc, HoodieActiveTimeline.createNewInstantTime());
+ table.rollbackBootstrap(context, HoodieActiveTimeline.createNewInstantTime());
LOG.info("Finished rolling back pending bootstrap");
}
@@ -181,21 +255,11 @@ public class HoodieWriteClient extends AbstractHo
/**
* Upsert a batch of new records into Hoodie table at the supplied instantTime.
*
- * @param records JavaRDD of hoodieRecords to upsert
+ * @param records hoodieRecords to upsert
* @param instantTime Instant time of the commit
- * @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
+ * @return WriteStatus to inspect errors and counts
*/
- public JavaRDD upsert(JavaRDD> records, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.UPSERT, instantTime);
- table.validateUpsertSchema();
- setOperationType(WriteOperationType.UPSERT);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.upsert(jsc, instantTime, records);
- if (result.getIndexLookupDuration().isPresent()) {
- metrics.updateIndexMetrics(LOOKUP_STR, result.getIndexLookupDuration().get().toMillis());
- }
- return postWrite(result, instantTime, table);
- }
+ public abstract O upsert(I records, final String instantTime);
/**
* Upserts the given prepared records into the Hoodie table, at the supplied instantTime.
@@ -206,14 +270,7 @@ public class HoodieWriteClient extends AbstractHo
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD upsertPreppedRecords(JavaRDD> preppedRecords, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.UPSERT_PREPPED, instantTime);
- table.validateUpsertSchema();
- setOperationType(WriteOperationType.UPSERT_PREPPED);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.upsertPrepped(jsc,instantTime, preppedRecords);
- return postWrite(result, instantTime, table);
- }
+ public abstract O upsertPreppedRecords(I preppedRecords, final String instantTime);
/**
* Inserts the given HoodieRecords, into the table. This API is intended to be used for normal writes.
@@ -225,14 +282,7 @@ public class HoodieWriteClient extends AbstractHo
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD insert(JavaRDD> records, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.INSERT, instantTime);
- table.validateInsertSchema();
- setOperationType(WriteOperationType.INSERT);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.insert(jsc,instantTime, records);
- return postWrite(result, instantTime, table);
- }
+ public abstract O insert(I records, final String instantTime);
/**
* Inserts the given prepared records into the Hoodie table, at the supplied instantTime.
@@ -245,36 +295,27 @@ public class HoodieWriteClient extends AbstractHo
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD insertPreppedRecords(JavaRDD> preppedRecords, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.INSERT_PREPPED, instantTime);
- table.validateInsertSchema();
- setOperationType(WriteOperationType.INSERT_PREPPED);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.insertPrepped(jsc,instantTime, preppedRecords);
- return postWrite(result, instantTime, table);
- }
+ public abstract O insertPreppedRecords(I preppedRecords, final String instantTime);
/**
* Loads the given HoodieRecords, as inserts into the table. This is suitable for doing big bulk loads into a Hoodie
* table for the very first time (e.g: converting an existing table to Hoodie).
*
* This implementation uses sortBy (which does range partitioning based on reservoir sampling) and attempts to control
- * the numbers of files with less memory compared to the {@link HoodieWriteClient#insert(JavaRDD, String)}
+ * the numbers of files with less memory compared to the {@link AbstractHoodieWriteClient#insert(I, String)}
*
* @param records HoodieRecords to insert
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD bulkInsert(JavaRDD> records, final String instantTime) {
- return bulkInsert(records, instantTime, Option.empty());
- }
+ public abstract O bulkInsert(I records, final String instantTime);
/**
* Loads the given HoodieRecords, as inserts into the table. This is suitable for doing big bulk loads into a Hoodie
* table for the very first time (e.g: converting an existing table to Hoodie).
*
* This implementation uses sortBy (which does range partitioning based on reservoir sampling) and attempts to control
- * the numbers of files with less memory compared to the {@link HoodieWriteClient#insert(JavaRDD, String)}. Optionally
+ * the numbers of files with less memory compared to the {@link AbstractHoodieWriteClient#insert(I, String)}. Optionally
* it allows users to specify their own partitioner. If specified then it will be used for repartitioning records. See
* {@link BulkInsertPartitioner}.
*
@@ -284,15 +325,9 @@ public class HoodieWriteClient extends AbstractHo
* into hoodie.
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD bulkInsert(JavaRDD> records, final String instantTime,
- Option userDefinedBulkInsertPartitioner) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.BULK_INSERT, instantTime);
- table.validateInsertSchema();
- setOperationType(WriteOperationType.BULK_INSERT);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.bulkInsert(jsc,instantTime, records, userDefinedBulkInsertPartitioner);
- return postWrite(result, instantTime, table);
- }
+ public abstract O bulkInsert(I records, final String instantTime,
+ Option> userDefinedBulkInsertPartitioner);
+
/**
* Loads the given HoodieRecords, as inserts into the table. This is suitable for doing big bulk loads into a Hoodie
@@ -300,7 +335,7 @@ public class HoodieWriteClient extends AbstractHo
* duplicates if needed.
*
* This implementation uses sortBy (which does range partitioning based on reservoir sampling) and attempts to control
- * the numbers of files with less memory compared to the {@link HoodieWriteClient#insert(JavaRDD, String)}. Optionally
+ * the numbers of files with less memory compared to the {@link AbstractHoodieWriteClient#insert(I, String)}. Optionally
* it allows users to specify their own partitioner. If specified then it will be used for repartitioning records. See
* {@link BulkInsertPartitioner}.
*
@@ -310,31 +345,8 @@ public class HoodieWriteClient extends AbstractHo
* into hoodie.
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD bulkInsertPreppedRecords(JavaRDD> preppedRecords, final String instantTime,
- Option bulkInsertPartitioner) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.BULK_INSERT_PREPPED, instantTime);
- table.validateInsertSchema();
- setOperationType(WriteOperationType.BULK_INSERT_PREPPED);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.bulkInsertPrepped(jsc,instantTime, preppedRecords, bulkInsertPartitioner);
- return postWrite(result, instantTime, table);
- }
-
- /**
- * Removes all existing records from the partitions affected and inserts the given HoodieRecords, into the table.
-
- * @param records HoodieRecords to insert
- * @param instantTime Instant time of the commit
- * @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
- */
- public HoodieWriteResult insertOverwrite(JavaRDD> records, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.INSERT_OVERWRITE, instantTime);
- table.validateInsertSchema();
- setOperationType(WriteOperationType.INSERT_OVERWRITE);
- this.asyncCleanerService = AsyncCleanerService.startAsyncCleaningIfEnabled(this, instantTime);
- HoodieWriteMetadata result = table.insertOverwrite(jsc, instantTime, records);
- return new HoodieWriteResult(postWrite(result, instantTime, table), result.getPartitionToReplaceFileIds());
- }
+ public abstract O bulkInsertPreppedRecords(I preppedRecords, final String instantTime,
+ Option> bulkInsertPartitioner);
/**
* Deletes a list of {@link HoodieKey}s from the Hoodie table, at the supplied instantTime {@link HoodieKey}s will be
@@ -344,12 +356,7 @@ public class HoodieWriteClient extends AbstractHo
* @param instantTime Commit time handle
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD delete(JavaRDD keys, final String instantTime) {
- HoodieTable table = getTableAndInitCtx(WriteOperationType.DELETE, instantTime);
- setOperationType(WriteOperationType.DELETE);
- HoodieWriteMetadata result = table.delete(jsc,instantTime, keys);
- return postWrite(result, instantTime, table);
- }
+ public abstract O delete(K keys, final String instantTime);
/**
* Common method containing steps to be performed after write (upsert/insert/..) operations including auto-commit.
@@ -358,31 +365,21 @@ public class HoodieWriteClient extends AbstractHo
* @param hoodieTable Hoodie Table
* @return Write Status
*/
- private JavaRDD postWrite(HoodieWriteMetadata result, String instantTime, HoodieTable hoodieTable) {
- if (result.getIndexLookupDuration().isPresent()) {
- metrics.updateIndexMetrics(getOperationType().name(), result.getIndexUpdateDuration().get().toMillis());
- }
- if (result.isCommitted()) {
- // Perform post commit operations.
- if (result.getFinalizeDuration().isPresent()) {
- metrics.updateFinalizeWriteMetrics(result.getFinalizeDuration().get().toMillis(),
- result.getWriteStats().get().size());
- }
+ protected abstract O postWrite(HoodieWriteMetadata result, String instantTime, HoodieTable hoodieTable);
- postCommit(hoodieTable, result.getCommitMetadata().get(), instantTime, Option.empty());
-
- emitCommitMetrics(instantTime, result.getCommitMetadata().get(),
- hoodieTable.getMetaClient().getCommitActionType());
- }
- return result.getWriteStatuses();
- }
-
- @Override
- protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option> extraMetadata) {
+ /**
+ * Post Commit Hook. Derived classes use this method to perform post-commit processing
+ *
+ * @param table table to commit on
+ * @param metadata Commit Metadata corresponding to committed instant
+ * @param instantTime Instant Time
+ * @param extraMetadata Additional Metadata passed by user
+ */
+ protected void postCommit(HoodieTable table, HoodieCommitMetadata metadata, String instantTime, Option> extraMetadata) {
try {
// Delete the marker directory for the instant.
- new MarkerFiles(table, instantTime).quietDeleteMarkerDir(jsc, config.getMarkersDeleteParallelism());
+ new MarkerFiles(table, instantTime).quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
// Do an inline compaction if enabled
if (config.isInlineCompaction()) {
@@ -393,15 +390,15 @@ public class HoodieWriteClient extends AbstractHo
metadata.addMetadata(HoodieCompactionConfig.INLINE_COMPACT_PROP, "false");
}
// We cannot have unbounded commit files. Archive commits if we have to archive
- HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, hadoopConf);
- archiveLog.archiveIfRequired(jsc);
+ HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, table);
+ archiveLog.archiveIfRequired(context);
autoCleanOnCommit(instantTime);
} catch (IOException ioe) {
throw new HoodieIOException(ioe.getMessage(), ioe);
}
}
- private void runAnyPendingCompactions(HoodieTable table) {
+ protected void runAnyPendingCompactions(HoodieTable table) {
table.getActiveTimeline().getCommitsAndCompactionTimeline().filterPendingCompactionTimeline().getInstants()
.forEach(instant -> {
LOG.info("Running previously failed inflight compaction at instant " + instant);
@@ -411,9 +408,10 @@ public class HoodieWriteClient extends AbstractHo
/**
* Handle auto clean during commit.
+ *
* @param instantTime
*/
- private void autoCleanOnCommit(String instantTime) {
+ protected void autoCleanOnCommit(String instantTime) {
if (config.isAutoClean()) {
// Call clean to cleanup if there is anything to cleanup after the commit,
if (config.isAsyncClean()) {
@@ -434,7 +432,7 @@ public class HoodieWriteClient extends AbstractHo
* @param comment - Comment for the savepoint
*/
public void savepoint(String user, String comment) {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
if (table.getCompletedCommitsTimeline().empty()) {
throw new HoodieSavepointException("Could not savepoint. Commit timeline is empty");
}
@@ -458,8 +456,8 @@ public class HoodieWriteClient extends AbstractHo
* @param comment - Comment for the savepoint
*/
public void savepoint(String instantTime, String user, String comment) {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
- table.savepoint(jsc, instantTime, user, comment);
+ HoodieTable table = createTable(config, hadoopConf);
+ table.savepoint(context, instantTime, user, comment);
}
/**
@@ -470,7 +468,7 @@ public class HoodieWriteClient extends AbstractHo
* @return true if the savepoint was deleted successfully
*/
public void deleteSavepoint(String savepointTime) {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
SavepointHelpers.deleteSavepoint(table, savepointTime);
}
@@ -485,7 +483,7 @@ public class HoodieWriteClient extends AbstractHo
* @return true if the savepoint was restored to successfully
*/
public void restoreToSavepoint(String savepointTime) {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
SavepointHelpers.validateSavepointPresence(table, savepointTime);
restoreToInstant(savepointTime);
SavepointHelpers.validateSavepointRestore(table, savepointTime);
@@ -500,16 +498,16 @@ public class HoodieWriteClient extends AbstractHo
public boolean rollback(final String commitInstantTime) throws HoodieRollbackException {
LOG.info("Begin rollback of instant " + commitInstantTime);
final String rollbackInstantTime = HoodieActiveTimeline.createNewInstantTime();
- final Timer.Context context = this.metrics.getRollbackCtx();
+ final Timer.Context timerContext = this.metrics.getRollbackCtx();
try {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
Option commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstants()
- .filter(instant -> HoodieActiveTimeline.EQUALS.test(instant.getTimestamp(), commitInstantTime))
- .findFirst());
+ .filter(instant -> HoodieActiveTimeline.EQUALS.test(instant.getTimestamp(), commitInstantTime))
+ .findFirst());
if (commitInstantOpt.isPresent()) {
- HoodieRollbackMetadata rollbackMetadata = table.rollback(jsc, rollbackInstantTime, commitInstantOpt.get(), true);
- if (context != null) {
- long durationInMs = metrics.getDurationInMs(context.stop());
+ HoodieRollbackMetadata rollbackMetadata = table.rollback(context, rollbackInstantTime, commitInstantOpt.get(), true);
+ if (timerContext != null) {
+ long durationInMs = metrics.getDurationInMs(timerContext.stop());
metrics.updateRollbackMetrics(durationInMs, rollbackMetadata.getTotalFilesDeleted());
}
return true;
@@ -531,12 +529,12 @@ public class HoodieWriteClient extends AbstractHo
public HoodieRestoreMetadata restoreToInstant(final String instantTime) throws HoodieRestoreException {
LOG.info("Begin restore to instant " + instantTime);
final String restoreInstantTime = HoodieActiveTimeline.createNewInstantTime();
- Timer.Context context = metrics.getRollbackCtx();
+ Timer.Context timerContext = metrics.getRollbackCtx();
try {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
- HoodieRestoreMetadata restoreMetadata = table.restore(jsc, restoreInstantTime, instantTime);
- if (context != null) {
- final long durationInMs = metrics.getDurationInMs(context.stop());
+ HoodieTable table = createTable(config, hadoopConf);
+ HoodieRestoreMetadata restoreMetadata = table.restore(context, restoreInstantTime, instantTime);
+ if (timerContext != null) {
+ final long durationInMs = metrics.getDurationInMs(timerContext.stop());
final long totalFilesDeleted = restoreMetadata.getHoodieRestoreMetadata().values().stream()
.flatMap(Collection::stream)
.mapToLong(HoodieRollbackMetadata::getTotalFilesDeleted)
@@ -549,16 +547,6 @@ public class HoodieWriteClient extends AbstractHo
}
}
- /**
- * Releases any resources used by the client.
- */
- @Override
- public void close() {
- AsyncCleanerService.forceShutdown(asyncCleanerService);
- asyncCleanerService = null;
- super.close();
- }
-
/**
* Clean up any stale/old files/data lying around (either on file storage or index storage) based on the
* configurations and CleaningPolicy used. (typically files that no longer can be used by a running query can be
@@ -566,10 +554,10 @@ public class HoodieWriteClient extends AbstractHo
*/
public HoodieCleanMetadata clean(String cleanInstantTime) throws HoodieIOException {
LOG.info("Cleaner started");
- final Timer.Context context = metrics.getCleanCtx();
- HoodieCleanMetadata metadata = HoodieTable.create(config, hadoopConf).clean(jsc, cleanInstantTime);
- if (context != null && metadata != null) {
- long durationMs = metrics.getDurationInMs(context.stop());
+ final Timer.Context timerContext = metrics.getCleanCtx();
+ HoodieCleanMetadata metadata = createTable(config, hadoopConf).clean(context, cleanInstantTime);
+ if (timerContext != null && metadata != null) {
+ long durationMs = metrics.getDurationInMs(timerContext.stop());
metrics.updateCleanMetrics(durationMs, metadata.getTotalFilesDeleted());
LOG.info("Cleaned " + metadata.getTotalFilesDeleted() + " files"
+ " Earliest Retained Instant :" + metadata.getEarliestCommitToRetain()
@@ -634,7 +622,7 @@ public class HoodieWriteClient extends AbstractHo
HoodieTimeline.compareTimestamps(latestPending.getTimestamp(), HoodieTimeline.LESSER_THAN, instantTime),
"Latest pending compaction instant time must be earlier than this instant time. Latest Compaction :"
+ latestPending + ", Ingesting at " + instantTime));
- metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(State.REQUESTED, actionType,
+ metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(HoodieInstant.State.REQUESTED, actionType,
instantTime));
}
@@ -656,8 +644,8 @@ public class HoodieWriteClient extends AbstractHo
*/
public boolean scheduleCompactionAtInstant(String instantTime, Option> extraMetadata) throws HoodieIOException {
LOG.info("Scheduling compaction at instant time :" + instantTime);
- Option plan = HoodieTable.create(config, hadoopConf)
- .scheduleCompaction(jsc, instantTime, extraMetadata);
+ Option plan = createTable(config, hadoopConf)
+ .scheduleCompaction(context, instantTime, extraMetadata);
return plan.isPresent();
}
@@ -667,7 +655,7 @@ public class HoodieWriteClient extends AbstractHo
* @param compactionInstantTime Compaction Instant Time
* @return RDD of WriteStatus to inspect errors and counts
*/
- public JavaRDD compact(String compactionInstantTime) {
+ public O compact(String compactionInstantTime) {
return compact(compactionInstantTime, config.shouldAutoCommit());
}
@@ -678,38 +666,15 @@ public class HoodieWriteClient extends AbstractHo
* @param writeStatuses RDD of WriteStatus to inspect errors and counts
* @param extraMetadata Extra Metadata to be stored
*/
- public void commitCompaction(String compactionInstantTime, JavaRDD writeStatuses,
- Option> extraMetadata) throws IOException {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
- HoodieCommitMetadata metadata = CompactHelpers.createCompactionMetadata(
- table, compactionInstantTime, writeStatuses, config.getSchema());
- extraMetadata.ifPresent(m -> m.forEach(metadata::addMetadata));
- completeCompaction(metadata, writeStatuses, table, compactionInstantTime);
- }
+ public abstract void commitCompaction(String compactionInstantTime, O writeStatuses,
+ Option> extraMetadata) throws IOException;
/**
* Commit Compaction and track metrics.
*/
- protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD writeStatuses, HoodieTable table,
- String compactionCommitTime) {
+ protected abstract void completeCompaction(HoodieCommitMetadata metadata, O writeStatuses,
+ HoodieTable table, String compactionCommitTime);
- List writeStats = writeStatuses.map(WriteStatus::getStat).collect();
- finalizeWrite(table, compactionCommitTime, writeStats);
- LOG.info("Committing Compaction " + compactionCommitTime + ". Finished with result " + metadata);
- CompactHelpers.completeInflightCompaction(table, compactionCommitTime, metadata);
-
- if (compactionTimer != null) {
- long durationInMs = metrics.getDurationInMs(compactionTimer.stop());
- try {
- metrics.updateCommitMetrics(HoodieActiveTimeline.COMMIT_FORMATTER.parse(compactionCommitTime).getTime(),
- durationInMs, metadata, HoodieActiveTimeline.COMPACTION_ACTION);
- } catch (ParseException e) {
- throw new HoodieCommitException("Commit time is not of valid format. Failed to commit compaction "
- + config.getBasePath() + " at time " + compactionCommitTime, e);
- }
- }
- LOG.info("Compacted successfully on commit " + compactionCommitTime);
- }
/**
* Rollback failed compactions. Inflight rollbacks for compactions revert the .inflight file to the .requested file
@@ -717,8 +682,8 @@ public class HoodieWriteClient extends AbstractHo
* @param inflightInstant Inflight Compaction Instant
* @param table Hoodie Table
*/
- public void rollbackInflightCompaction(HoodieInstant inflightInstant, HoodieTable table) {
- table.rollback(jsc, HoodieActiveTimeline.createNewInstantTime(), inflightInstant, false);
+ public void rollbackInflightCompaction(HoodieInstant inflightInstant, HoodieTable table) {
+ table.rollback(context, HoodieActiveTimeline.createNewInstantTime(), inflightInstant, false);
table.getActiveTimeline().revertCompactionInflightToRequested(inflightInstant);
}
@@ -726,7 +691,7 @@ public class HoodieWriteClient extends AbstractHo
* Cleanup all pending commits.
*/
private void rollbackPendingCommits() {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
+ HoodieTable table = createTable(config, hadoopConf);
HoodieTimeline inflightTimeline = table.getMetaClient().getCommitsTimeline().filterPendingExcludingCompaction();
List commits = inflightTimeline.getReverseOrderedInstants().map(HoodieInstant::getTimestamp)
.collect(Collectors.toList());
@@ -747,27 +712,12 @@ public class HoodieWriteClient extends AbstractHo
* @param compactionInstantTime Compaction Instant Time
* @return RDD of Write Status
*/
- private JavaRDD compact(String compactionInstantTime, boolean shouldComplete) {
- HoodieTable table = HoodieTable.create(config, hadoopConf);
- HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
- if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
- rollbackInflightCompaction(inflightInstant, table);
- table.getMetaClient().reloadActiveTimeline();
- }
- compactionTimer = metrics.getCompactionCtx();
- HoodieWriteMetadata compactionMetadata = table.compact(jsc, compactionInstantTime);
- JavaRDD statuses = compactionMetadata.getWriteStatuses();
- if (shouldComplete && compactionMetadata.getCommitMetadata().isPresent()) {
- completeCompaction(compactionMetadata.getCommitMetadata().get(), statuses, table, compactionInstantTime);
- }
- return statuses;
- }
+ protected abstract O compact(String compactionInstantTime, boolean shouldComplete);
/**
* Performs a compaction operation on a table, serially before or after an insert/upsert action.
*/
- private Option inlineCompact(Option> extraMetadata) {
+ protected Option inlineCompact(Option> extraMetadata) {
Option compactionInstantTimeOpt = scheduleCompaction(extraMetadata);
compactionInstantTimeOpt.ifPresent(compactionInstantTime -> {
// inline compaction should auto commit as the user is never given control
@@ -775,4 +725,82 @@ public class HoodieWriteClient extends AbstractHo
});
return compactionInstantTimeOpt;
}
+
+ /**
+ * Finalize Write operation.
+ *
+ * @param table HoodieTable
+ * @param instantTime Instant Time
+ * @param stats Hoodie Write Stat
+ */
+ protected void finalizeWrite(HoodieTable table, String instantTime, List stats) {
+ try {
+ final Timer.Context finalizeCtx = metrics.getFinalizeCtx();
+ table.finalizeWrite(context, instantTime, stats);
+ if (finalizeCtx != null) {
+ Option durationInMs = Option.of(metrics.getDurationInMs(finalizeCtx.stop()));
+ durationInMs.ifPresent(duration -> {
+ LOG.info("Finalize write elapsed time (milliseconds): " + duration);
+ metrics.updateFinalizeWriteMetrics(duration, stats.size());
+ });
+ }
+ } catch (HoodieIOException ioe) {
+ throw new HoodieCommitException("Failed to complete commit " + instantTime + " due to finalize errors.", ioe);
+ }
+ }
+
+ public HoodieMetrics getMetrics() {
+ return metrics;
+ }
+
+ public HoodieIndex getIndex() {
+ return index;
+ }
+
+ /**
+ * Get HoodieTable and init {@link Timer.Context}.
+ *
+ * @param operationType write operation type
+ * @param instantTime current inflight instant time
+ * @return HoodieTable
+ */
+ protected abstract HoodieTable getTableAndInitCtx(WriteOperationType operationType, String instantTime);
+
+ /**
+ * Sets write schema from last instant since deletes may not have schema set in the config.
+ */
+ protected void setWriteSchemaForDeletes(HoodieTableMetaClient metaClient) {
+ try {
+ HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+ Option lastInstant =
+ activeTimeline.filterCompletedInstants().filter(s -> s.getAction().equals(metaClient.getCommitActionType()))
+ .lastInstant();
+ if (lastInstant.isPresent()) {
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(
+ activeTimeline.getInstantDetails(lastInstant.get()).get(), HoodieCommitMetadata.class);
+ if (commitMetadata.getExtraMetadata().containsKey(HoodieCommitMetadata.SCHEMA_KEY)) {
+ config.setSchema(commitMetadata.getExtraMetadata().get(HoodieCommitMetadata.SCHEMA_KEY));
+ } else {
+ throw new HoodieIOException("Latest commit does not have any schema in commit metadata");
+ }
+ } else {
+ throw new HoodieIOException("Deletes issued without any prior commits");
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("IOException thrown while reading last commit metadata", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ // release AsyncCleanerService
+ AsyncCleanerService.forceShutdown(asyncCleanerService);
+ asyncCleanerService = null;
+
+ // Stop timeline-server if running
+ super.close();
+ // Calling this here releases any resources used by your index, so make sure to finish any related operations
+ // before this point
+ this.index.close();
+ }
}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/AsyncCleanerService.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AsyncCleanerService.java
similarity index 90%
rename from hudi-client/src/main/java/org/apache/hudi/client/AsyncCleanerService.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AsyncCleanerService.java
index 6367e7972..e8016c957 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/AsyncCleanerService.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/AsyncCleanerService.java
@@ -18,7 +18,7 @@
package org.apache.hudi.client;
-import org.apache.hudi.async.AbstractAsyncService;
+import org.apache.hudi.async.HoodieAsyncService;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.log4j.LogManager;
@@ -31,15 +31,15 @@ import java.util.concurrent.Executors;
/**
* Clean service running concurrently with write operation.
*/
-class AsyncCleanerService extends AbstractAsyncService {
+class AsyncCleanerService extends HoodieAsyncService {
private static final Logger LOG = LogManager.getLogger(AsyncCleanerService.class);
- private final HoodieWriteClient writeClient;
+ private final AbstractHoodieWriteClient writeClient;
private final String cleanInstantTime;
private final transient ExecutorService executor = Executors.newSingleThreadExecutor();
- protected AsyncCleanerService(HoodieWriteClient writeClient, String cleanInstantTime) {
+ protected AsyncCleanerService(AbstractHoodieWriteClient writeClient, String cleanInstantTime) {
this.writeClient = writeClient;
this.cleanInstantTime = cleanInstantTime;
}
@@ -52,7 +52,7 @@ class AsyncCleanerService extends AbstractAsyncService {
}), executor);
}
- public static AsyncCleanerService startAsyncCleaningIfEnabled(HoodieWriteClient writeClient,
+ public static AsyncCleanerService startAsyncCleaningIfEnabled(AbstractHoodieWriteClient writeClient,
String instantTime) {
AsyncCleanerService asyncCleanerService = null;
if (writeClient.getConfig().isAutoClean() && writeClient.getConfig().isAsyncClean()) {
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/CompactionAdminClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
similarity index 97%
rename from hudi-client/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
index 48042e72e..a2ecb6727 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
@@ -22,6 +22,7 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hudi.avro.model.HoodieCompactionOperation;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.CompactionOperation;
import org.apache.hudi.common.model.FileSlice;
@@ -45,7 +46,6 @@ import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.table.action.compact.OperationResult;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.spark.api.java.JavaSparkContext;
import java.io.FileNotFoundException;
import java.io.IOException;
@@ -65,8 +65,8 @@ public class CompactionAdminClient extends AbstractHoodieClient {
private static final Logger LOG = LogManager.getLogger(CompactionAdminClient.class);
- public CompactionAdminClient(JavaSparkContext jsc, String basePath) {
- super(jsc, HoodieWriteConfig.newBuilder().withPath(basePath).build());
+ public CompactionAdminClient(HoodieEngineContext context, String basePath) {
+ super(context, HoodieWriteConfig.newBuilder().withPath(basePath).build());
}
/**
@@ -85,14 +85,14 @@ public class CompactionAdminClient extends AbstractHoodieClient {
if (plan.getOperations() != null) {
List ops = plan.getOperations().stream()
.map(CompactionOperation::convertFromAvroRecordInstance).collect(Collectors.toList());
- jsc.setJobGroup(this.getClass().getSimpleName(), "Validate compaction operations");
- return jsc.parallelize(ops, parallelism).map(op -> {
+ context.setJobStatus(this.getClass().getSimpleName(), "Validate compaction operations");
+ return context.map(ops, op -> {
try {
return validateCompactionOperation(metaClient, compactionInstant, op, Option.of(fsView));
} catch (IOException e) {
throw new HoodieIOException(e.getMessage(), e);
}
- }).collect();
+ }, parallelism);
}
return new ArrayList<>();
}
@@ -351,8 +351,8 @@ public class CompactionAdminClient extends AbstractHoodieClient {
} else {
LOG.info("The following compaction renaming operations needs to be performed to un-schedule");
if (!dryRun) {
- jsc.setJobGroup(this.getClass().getSimpleName(), "Execute unschedule operations");
- return jsc.parallelize(renameActions, parallelism).map(lfPair -> {
+ context.setJobStatus(this.getClass().getSimpleName(), "Execute unschedule operations");
+ return context.map(renameActions, lfPair -> {
try {
LOG.info("RENAME " + lfPair.getLeft().getPath() + " => " + lfPair.getRight().getPath());
renameLogFile(metaClient, lfPair.getLeft(), lfPair.getRight());
@@ -363,7 +363,7 @@ public class CompactionAdminClient extends AbstractHoodieClient {
+ lfPair.getLeft().getBaseCommitTime() + "\" to recover from failure ***\n\n\n");
return new RenameOpResult(lfPair, false, Option.of(e));
}
- }).collect();
+ }, parallelism);
} else {
LOG.info("Dry-Run Mode activated for rename operations");
return renameActions.parallelStream().map(lfPair -> new RenameOpResult(lfPair, false, false, Option.empty()))
@@ -394,17 +394,17 @@ public class CompactionAdminClient extends AbstractHoodieClient {
"Number of Compaction Operations :" + plan.getOperations().size() + " for instant :" + compactionInstant);
List ops = plan.getOperations().stream()
.map(CompactionOperation::convertFromAvroRecordInstance).collect(Collectors.toList());
- jsc.setJobGroup(this.getClass().getSimpleName(), "Generate compaction unscheduling operations");
- return jsc.parallelize(ops, parallelism).flatMap(op -> {
+ context.setJobStatus(this.getClass().getSimpleName(), "Generate compaction unscheduling operations");
+ return context.flatMap(ops, op -> {
try {
return getRenamingActionsForUnschedulingCompactionOperation(metaClient, compactionInstant, op,
- Option.of(fsView), skipValidation).iterator();
+ Option.of(fsView), skipValidation).stream();
} catch (IOException ioe) {
throw new HoodieIOException(ioe.getMessage(), ioe);
} catch (CompactionValidationException ve) {
throw new HoodieException(ve);
}
- }).collect();
+ }, parallelism);
}
LOG.warn("No operations for compaction instant : " + compactionInstant);
return new ArrayList<>();
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/HoodieInternalWriteStatus.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/table/ReplaceArchivalHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/ReplaceArchivalHelper.java
similarity index 87%
rename from hudi-client/src/main/java/org/apache/hudi/table/ReplaceArchivalHelper.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/ReplaceArchivalHelper.java
index 317cf0de1..515f43e64 100644
--- a/hudi-client/src/main/java/org/apache/hudi/table/ReplaceArchivalHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/ReplaceArchivalHelper.java
@@ -16,21 +16,23 @@
* limitations under the License.
*/
-package org.apache.hudi.table;
+package org.apache.hudi.client;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.hadoop.fs.Path;
+
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
import org.apache.hudi.common.model.HoodieRollingStatMetadata;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.view.TableFileSystemView;
+import org.apache.hudi.client.common.HoodieEngineContext;
+
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
import java.io.IOException;
import java.io.Serializable;
@@ -63,17 +65,17 @@ public class ReplaceArchivalHelper implements Serializable {
/**
* Delete all files represented by FileSlices in parallel. Return true if all files are deleted successfully.
*/
- public static boolean deleteReplacedFileGroups(JavaSparkContext jsc, HoodieTableMetaClient metaClient,
+ public static boolean deleteReplacedFileGroups(HoodieEngineContext context, HoodieTableMetaClient metaClient,
TableFileSystemView fileSystemView,
HoodieInstant instant, List replacedPartitions) {
- JavaRDD partitions = jsc.parallelize(replacedPartitions, replacedPartitions.size());
- return partitions.map(partition -> {
+ List f = context.map(replacedPartitions, partition -> {
Stream fileSlices = fileSystemView.getReplacedFileGroupsBeforeOrOn(instant.getTimestamp(), partition)
- .flatMap(g -> g.getAllRawFileSlices());
+ .flatMap(HoodieFileGroup::getAllRawFileSlices);
+ return fileSlices.allMatch(slice -> deleteFileSlice(slice, metaClient, instant));
+ }, replacedPartitions.size());
- return fileSlices.map(slice -> deleteFileSlice(slice, metaClient, instant)).allMatch(x -> x);
- }).reduce((x, y) -> x & y);
+ return f.stream().reduce((x, y) -> x & y).orElse(true);
}
private static boolean deleteFileSlice(FileSlice fileSlice, HoodieTableMetaClient metaClient, HoodieInstant instant) {
@@ -95,5 +97,4 @@ public class ReplaceArchivalHelper implements Serializable {
return false;
}
}
-
}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/WriteStatus.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/WriteStatus.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapMode.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapMode.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapMode.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapMode.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapRecordPayload.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapRecordPayload.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapRecordPayload.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapRecordPayload.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapWriteStatus.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapWriteStatus.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/BootstrapWriteStatus.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/BootstrapWriteStatus.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java
similarity index 75%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java
index 8b077ad9f..542dad9cc 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/FullRecordBootstrapDataProvider.java
@@ -20,30 +20,28 @@ package org.apache.hudi.client.bootstrap;
import java.io.Serializable;
import org.apache.hudi.avro.model.HoodieFileStatus;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
import java.util.List;
/**
- * Creates RDD of Hoodie Records with complete record data, given a list of partitions to be bootstrapped.
+ * Creates Hoodie Records with complete record data, given a list of partitions to be bootstrapped.
*/
-public abstract class FullRecordBootstrapDataProvider implements Serializable {
+public abstract class FullRecordBootstrapDataProvider implements Serializable {
protected static final Logger LOG = LogManager.getLogger(FullRecordBootstrapDataProvider.class);
protected final TypedProperties props;
- protected final transient JavaSparkContext jsc;
+ protected final transient HoodieEngineContext context;
- public FullRecordBootstrapDataProvider(TypedProperties props, JavaSparkContext jsc) {
+ public FullRecordBootstrapDataProvider(TypedProperties props, HoodieEngineContext context) {
this.props = props;
- this.jsc = jsc;
+ this.context = context;
}
/**
@@ -51,8 +49,8 @@ public abstract class FullRecordBootstrapDataProvider implements Serializable {
* @param tableName Hudi Table Name
* @param sourceBasePath Source Base Path
* @param partitionPaths Partition Paths
- * @return JavaRDD of input records
+ * @return input records
*/
- public abstract JavaRDD generateInputRecordRDD(String tableName,
+ public abstract I generateInputRecords(String tableName,
String sourceBasePath, List>> partitionPaths);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/HoodieBootstrapSchemaProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/HoodieBootstrapSchemaProvider.java
new file mode 100644
index 000000000..f63345d91
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/HoodieBootstrapSchemaProvider.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.bootstrap;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.avro.model.HoodieFileStatus;
+import org.apache.hudi.client.common.HoodieEngineContext;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.apache.avro.Schema;
+
+import java.util.List;
+
+/**
+ * Bootstrap Schema Provider. Schema provided in config is used. If not available, use schema from Parquet
+ */
+public abstract class HoodieBootstrapSchemaProvider {
+
+ protected final HoodieWriteConfig writeConfig;
+
+ public HoodieBootstrapSchemaProvider(HoodieWriteConfig writeConfig) {
+ this.writeConfig = writeConfig;
+ }
+
+ /**
+ * Main API to select avro schema for bootstrapping.
+ * @param context HoodieEngineContext
+ * @param partitions List of partitions with files within them
+ * @return Avro Schema
+ */
+ public final Schema getBootstrapSchema(HoodieEngineContext context, List>> partitions) {
+ if (writeConfig.getSchema() != null) {
+ // Use schema specified by user if set
+ Schema userSchema = Schema.parse(writeConfig.getSchema());
+ if (!HoodieAvroUtils.getNullSchema().equals(userSchema)) {
+ return userSchema;
+ }
+ }
+ return getBootstrapSourceSchema(context, partitions);
+ }
+
+ /**
+ * Select a random file to be used to generate avro schema.
+ * Override this method to get custom schema selection.
+ * @param context HoodieEngineContext
+ * @param partitions List of partitions with files within them
+ * @return Avro Schema
+ */
+ protected abstract Schema getBootstrapSourceSchema(HoodieEngineContext context,
+ List>> partitions);
+
+}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapModeSelector.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapModeSelector.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapModeSelector.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/BootstrapRegexModeSelector.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/FullRecordBootstrapModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/FullRecordBootstrapModeSelector.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/FullRecordBootstrapModeSelector.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/FullRecordBootstrapModeSelector.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/MetadataOnlyBootstrapModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/MetadataOnlyBootstrapModeSelector.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/MetadataOnlyBootstrapModeSelector.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/MetadataOnlyBootstrapModeSelector.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/UniformBootstrapModeSelector.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/UniformBootstrapModeSelector.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/selector/UniformBootstrapModeSelector.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/selector/UniformBootstrapModeSelector.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/translator/BootstrapPartitionPathTranslator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/translator/BootstrapPartitionPathTranslator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/translator/BootstrapPartitionPathTranslator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/translator/BootstrapPartitionPathTranslator.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/bootstrap/translator/IdentityBootstrapPartitionPathTranslator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/translator/IdentityBootstrapPartitionPathTranslator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/bootstrap/translator/IdentityBootstrapPartitionPathTranslator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/bootstrap/translator/IdentityBootstrapPartitionPathTranslator.java
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/EngineProperty.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/EngineProperty.java
new file mode 100644
index 000000000..aeaec3274
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/EngineProperty.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.common;
+
+/**
+ * Properties specific to each engine, that can be set/obtained from.
+ */
+public enum EngineProperty {
+ // hostname to bind embedded timeline server to
+ EMBEDDED_SERVER_HOST,
+ // Pool/queue to use to run compaction.
+ COMPACTION_POOL_NAME,
+ // Amount of total memory available to each engine executor
+ TOTAL_MEMORY_AVAILABLE,
+ // Fraction of that memory, that is already in use by the engine
+ MEMORY_FRACTION_IN_USE,
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/HoodieEngineContext.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/HoodieEngineContext.java
new file mode 100644
index 000000000..f8d889640
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/HoodieEngineContext.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.common;
+
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.client.common.function.SerializableConsumer;
+import org.apache.hudi.client.common.function.SerializableFunction;
+import org.apache.hudi.client.common.function.SerializablePairFunction;
+import org.apache.hudi.common.util.Option;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+/**
+ * Base class contains the context information needed by the engine at runtime. It will be extended by different
+ * engine implementation if needed.
+ */
+public abstract class HoodieEngineContext {
+
+ /**
+ * A wrapped hadoop configuration which can be serialized.
+ */
+ private SerializableConfiguration hadoopConf;
+
+ private TaskContextSupplier taskContextSupplier;
+
+ public HoodieEngineContext(SerializableConfiguration hadoopConf, TaskContextSupplier taskContextSupplier) {
+ this.hadoopConf = hadoopConf;
+ this.taskContextSupplier = taskContextSupplier;
+ }
+
+ public SerializableConfiguration getHadoopConf() {
+ return hadoopConf;
+ }
+
+ public TaskContextSupplier getTaskContextSupplier() {
+ return taskContextSupplier;
+ }
+
+ public abstract List map(List data, SerializableFunction func, int parallelism);
+
+ public abstract List flatMap(List data, SerializableFunction> func, int parallelism);
+
+ public abstract void foreach(List data, SerializableConsumer consumer, int parallelism);
+
+ public abstract Map mapToPair(List data, SerializablePairFunction func, Integer parallelism);
+
+ public abstract void setProperty(EngineProperty key, String value);
+
+ public abstract Option getProperty(EngineProperty key);
+
+ public abstract void setJobStatus(String activeModule, String activityDescription);
+
+}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/SparkTaskContextSupplier.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/TaskContextSupplier.java
similarity index 65%
rename from hudi-client/src/main/java/org/apache/hudi/client/SparkTaskContextSupplier.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/TaskContextSupplier.java
index 601dd98a2..3a350d61d 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/SparkTaskContextSupplier.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/TaskContextSupplier.java
@@ -16,27 +16,23 @@
* limitations under the License.
*/
-package org.apache.hudi.client;
+package org.apache.hudi.client.common;
-import org.apache.spark.TaskContext;
+import org.apache.hudi.common.util.Option;
import java.io.Serializable;
import java.util.function.Supplier;
/**
- * Spark task context supplier.
+ * Base task context supplier.
*/
-public class SparkTaskContextSupplier implements Serializable {
+public abstract class TaskContextSupplier implements Serializable {
- public Supplier getPartitionIdSupplier() {
- return () -> TaskContext.getPartitionId();
- }
+ public abstract Supplier getPartitionIdSupplier();
- public Supplier getStageIdSupplier() {
- return () -> TaskContext.get().stageId();
- }
+ public abstract Supplier getStageIdSupplier();
- public Supplier getAttemptIdSupplier() {
- return () -> TaskContext.get().taskAttemptId();
- }
+ public abstract Supplier getAttemptIdSupplier();
+
+ public abstract Option getProperty(EngineProperty prop);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableConsumer.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableConsumer.java
new file mode 100644
index 000000000..d7c420522
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableConsumer.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.common.function;
+
+import java.io.Serializable;
+
+/**
+ * A wrapped {@link java.util.function.Consumer} which can be serialized.
+ *
+ * @param input type
+ */
+@FunctionalInterface
+public interface SerializableConsumer extends Serializable {
+ void accept(I t) throws Exception;
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableFunction.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableFunction.java
new file mode 100644
index 000000000..d3714bc5b
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializableFunction.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.common.function;
+
+import java.io.Serializable;
+
+/**
+ * A wrapped {@link java.util.function.Function} which can be serialized.
+ *
+ * @param input data type
+ * @param output data type
+ */
+@FunctionalInterface
+public interface SerializableFunction extends Serializable {
+ O apply(I v1) throws Exception;
+}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializablePairFunction.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializablePairFunction.java
new file mode 100644
index 000000000..155837b7f
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/common/function/SerializablePairFunction.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.client.common.function;
+
+import scala.Tuple2;
+
+import java.io.Serializable;
+
+/**
+ * A function that returns key-value pairs (Tuple2<K, V>).
+ */
+@FunctionalInterface
+public interface SerializablePairFunction extends Serializable {
+ Tuple2 call(I t) throws Exception;
+}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
similarity index 85%
rename from hudi-client/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
index 730585b12..ca0c71350 100644
--- a/hudi-client/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java
@@ -18,6 +18,7 @@
package org.apache.hudi.client.embedded;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.config.SerializableConfiguration;
import org.apache.hudi.common.table.view.FileSystemViewManager;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
@@ -25,10 +26,8 @@ import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.common.util.NetworkUtils;
import org.apache.hudi.timeline.service.TimelineService;
-import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
-import org.apache.spark.SparkConf;
import java.io.IOException;
@@ -46,13 +45,10 @@ public class EmbeddedTimelineService {
private transient FileSystemViewManager viewManager;
private transient TimelineService server;
- public EmbeddedTimelineService(Configuration hadoopConf, SparkConf sparkConf, FileSystemViewStorageConfig config) {
- setHostAddrFromSparkConf(sparkConf);
- if (hostAddr == null) {
- this.hostAddr = NetworkUtils.getHostname();
- }
+ public EmbeddedTimelineService(HoodieEngineContext context, String embeddedTimelineServiceHostAddr, FileSystemViewStorageConfig config) {
+ setHostAddr(embeddedTimelineServiceHostAddr);
this.config = config;
- this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this.hadoopConf = context.getHadoopConf();
this.viewManager = createViewManager();
}
@@ -75,13 +71,13 @@ public class EmbeddedTimelineService {
LOG.info("Started embedded timeline server at " + hostAddr + ":" + serverPort);
}
- private void setHostAddrFromSparkConf(SparkConf sparkConf) {
- String hostAddr = sparkConf.get("spark.driver.host", null);
- if (hostAddr != null) {
- LOG.info("Overriding hostIp to (" + hostAddr + ") found in spark-conf. It was " + this.hostAddr);
- this.hostAddr = hostAddr;
+ private void setHostAddr(String embeddedTimelineServiceHostAddr) {
+ if (embeddedTimelineServiceHostAddr != null) {
+ LOG.info("Overriding hostIp to (" + embeddedTimelineServiceHostAddr + ") found in spark-conf. It was " + this.hostAddr);
+ this.hostAddr = embeddedTimelineServiceHostAddr;
} else {
LOG.warn("Unable to find driver bind address from spark config");
+ this.hostAddr = NetworkUtils.getHostname();
}
}
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LazyIterableIterator.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/utils/MergingIterator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MergingIterator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/client/utils/MergingIterator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MergingIterator.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieBootstrapConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieHBaseIndexConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieIndexConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java
similarity index 96%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java
index 3105fcd13..687033f45 100644
--- a/hudi-client/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMemoryConfig.java
@@ -118,6 +118,8 @@ public class HoodieMemoryConfig extends DefaultHoodieConfig {
String.valueOf(DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE));
setDefaultOnCondition(props, !props.containsKey(SPILLABLE_MAP_BASE_PATH_PROP), SPILLABLE_MAP_BASE_PATH_PROP,
DEFAULT_SPILLABLE_MAP_BASE_PATH);
+ setDefaultOnCondition(props, !props.containsKey(MAX_MEMORY_FOR_MERGE_PROP), MAX_MEMORY_FOR_MERGE_PROP,
+ String.valueOf(DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES));
setDefaultOnCondition(props, !props.containsKey(WRITESTATUS_FAILURE_FRACTION_PROP),
WRITESTATUS_FAILURE_FRACTION_PROP, String.valueOf(DEFAULT_WRITESTATUS_FAILURE_FRACTION));
return config;
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsDatadogConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsDatadogConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsDatadogConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsDatadogConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsPrometheusConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsPrometheusConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieMetricsPrometheusConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieMetricsPrometheusConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteCommitCallbackConfig.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
similarity index 99%
rename from hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 089474d15..60ca3c2dd 100644
--- a/hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -19,7 +19,6 @@
package org.apache.hudi.config;
import org.apache.hadoop.hbase.io.compress.Compression;
-import org.apache.hudi.client.HoodieWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.bootstrap.BootstrapMode;
import org.apache.hudi.common.config.DefaultHoodieConfig;
@@ -51,7 +50,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
/**
- * Class storing configs for the {@link HoodieWriteClient}.
+ * Class storing configs for the HoodieWriteClient.
*/
@Immutable
public class HoodieWriteConfig extends DefaultHoodieConfig {
@@ -672,7 +671,7 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
public boolean getPushGatewayRandomJobNameSuffix() {
return Boolean.parseBoolean(props.getProperty(HoodieMetricsPrometheusConfig.PUSHGATEWAY_RANDOM_JOB_NAME_SUFFIX));
}
-
+
/**
* memory configs.
*/
@@ -755,6 +754,10 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
return Integer.parseInt(props.getProperty(HoodieBootstrapConfig.BOOTSTRAP_PARALLELISM));
}
+ public Long getMaxMemoryPerPartitionMerge() {
+ return Long.valueOf(props.getProperty(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE_PROP));
+ }
+
public static class Builder {
protected final Properties props = new Properties();
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieAppendException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieAppendException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieAppendException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieAppendException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieCommitCallbackException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCommitCallbackException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieCommitCallbackException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCommitCallbackException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieCommitException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCommitException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieCommitException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCommitException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieCompactionException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCompactionException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieCompactionException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieCompactionException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieDependentSystemUnavailableException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieDependentSystemUnavailableException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieDependentSystemUnavailableException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieDependentSystemUnavailableException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieInsertException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieInsertException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieInsertException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieInsertException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieRestoreException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieRestoreException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieRestoreException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieRestoreException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieRollbackException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieRollbackException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieRollbackException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieRollbackException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieSavepointException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieSavepointException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieSavepointException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieSavepointException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieUpgradeDowngradeException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieUpgradeDowngradeException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieUpgradeDowngradeException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieUpgradeDowngradeException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieUpsertException.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieUpsertException.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/exception/HoodieUpsertException.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/exception/HoodieUpsertException.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java
similarity index 84%
rename from hudi-client/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java
index ae61d8068..8af72f351 100644
--- a/hudi-client/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/CopyOnWriteInsertHandler.java
@@ -18,13 +18,13 @@
package org.apache.hudi.execution;
-import org.apache.hudi.client.SparkTaskContextSupplier;
import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.util.queue.BoundedInMemoryQueueConsumer;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.execution.LazyInsertIterable.HoodieInsertValueGenResult;
+import org.apache.hudi.execution.HoodieLazyInsertIterable.HoodieInsertValueGenResult;
import org.apache.hudi.io.HoodieWriteHandle;
import org.apache.hudi.io.WriteHandleFactory;
import org.apache.hudi.table.HoodieTable;
@@ -43,10 +43,10 @@ public class CopyOnWriteInsertHandler
private HoodieWriteConfig config;
private String instantTime;
private boolean areRecordsSorted;
- private HoodieTable hoodieTable;
+ private HoodieTable hoodieTable;
private String idPrefix;
- private SparkTaskContextSupplier sparkTaskContextSupplier;
- private WriteHandleFactory writeHandleFactory;
+ private TaskContextSupplier taskContextSupplier;
+ private WriteHandleFactory writeHandleFactory;
private final List statuses = new ArrayList<>();
// Stores the open HoodieWriteHandle for each table partition path
@@ -55,15 +55,15 @@ public class CopyOnWriteInsertHandler
private Map handles = new HashMap<>();
public CopyOnWriteInsertHandler(HoodieWriteConfig config, String instantTime,
- boolean areRecordsSorted, HoodieTable hoodieTable, String idPrefix,
- SparkTaskContextSupplier sparkTaskContextSupplier,
- WriteHandleFactory writeHandleFactory) {
+ boolean areRecordsSorted, HoodieTable hoodieTable, String idPrefix,
+ TaskContextSupplier taskContextSupplier,
+ WriteHandleFactory writeHandleFactory) {
this.config = config;
this.instantTime = instantTime;
this.areRecordsSorted = areRecordsSorted;
this.hoodieTable = hoodieTable;
this.idPrefix = idPrefix;
- this.sparkTaskContextSupplier = sparkTaskContextSupplier;
+ this.taskContextSupplier = taskContextSupplier;
this.writeHandleFactory = writeHandleFactory;
}
@@ -81,7 +81,7 @@ public class CopyOnWriteInsertHandler
}
// Lazily initialize the handle, for the first time
handle = writeHandleFactory.create(config, instantTime, hoodieTable,
- insertPayload.getPartitionPath(), idPrefix, sparkTaskContextSupplier);
+ insertPayload.getPartitionPath(), idPrefix, taskContextSupplier);
handles.put(partitionPath, handle);
}
@@ -90,7 +90,7 @@ public class CopyOnWriteInsertHandler
statuses.add(handle.close());
// Open new handle
handle = writeHandleFactory.create(config, instantTime, hoodieTable,
- insertPayload.getPartitionPath(), idPrefix, sparkTaskContextSupplier);
+ insertPayload.getPartitionPath(), idPrefix, taskContextSupplier);
handles.put(partitionPath, handle);
}
handle.write(insertPayload, payload.insertValue, payload.exception);
diff --git a/hudi-client/src/main/java/org/apache/hudi/execution/LazyInsertIterable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
similarity index 58%
rename from hudi-client/src/main/java/org/apache/hudi/execution/LazyInsertIterable.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
index 572956d27..b435c68de 100644
--- a/hudi-client/src/main/java/org/apache/hudi/execution/LazyInsertIterable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/HoodieLazyInsertIterable.java
@@ -18,15 +18,13 @@
package org.apache.hudi.execution;
-import org.apache.hudi.client.SparkTaskContextSupplier;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.utils.LazyIterableIterator;
+import org.apache.hudi.client.common.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.queue.BoundedInMemoryExecutor;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.io.CreateHandleFactory;
import org.apache.hudi.io.WriteHandleFactory;
import org.apache.hudi.table.HoodieTable;
@@ -41,41 +39,39 @@ import java.util.function.Function;
/**
* Lazy Iterable, that writes a stream of HoodieRecords sorted by the partitionPath, into new files.
*/
-public class LazyInsertIterable
+public abstract class HoodieLazyInsertIterable
extends LazyIterableIterator, List> {
protected final HoodieWriteConfig hoodieConfig;
protected final String instantTime;
protected boolean areRecordsSorted;
- protected final HoodieTable hoodieTable;
+ protected final HoodieTable hoodieTable;
protected final String idPrefix;
- protected SparkTaskContextSupplier sparkTaskContextSupplier;
- protected WriteHandleFactory writeHandleFactory;
+ protected TaskContextSupplier taskContextSupplier;
+ protected WriteHandleFactory writeHandleFactory;
- public LazyInsertIterable(Iterator> sortedRecordItr, HoodieWriteConfig config,
- String instantTime, HoodieTable hoodieTable, String idPrefix,
- SparkTaskContextSupplier sparkTaskContextSupplier) {
- this(sortedRecordItr, true, config, instantTime, hoodieTable, idPrefix, sparkTaskContextSupplier);
- }
-
- public LazyInsertIterable(Iterator> recordItr, boolean areRecordsSorted,
- HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable,
- String idPrefix, SparkTaskContextSupplier sparkTaskContextSupplier) {
- this(recordItr, areRecordsSorted, config, instantTime, hoodieTable, idPrefix, sparkTaskContextSupplier,
+ public HoodieLazyInsertIterable(Iterator> recordItr,
+ boolean areRecordsSorted,
+ HoodieWriteConfig config,
+ String instantTime,
+ HoodieTable hoodieTable,
+ String idPrefix,
+ TaskContextSupplier taskContextSupplier) {
+ this(recordItr, areRecordsSorted, config, instantTime, hoodieTable, idPrefix, taskContextSupplier,
new CreateHandleFactory<>());
}
- public LazyInsertIterable(Iterator> recordItr, boolean areRecordsSorted,
- HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable,
- String idPrefix, SparkTaskContextSupplier sparkTaskContextSupplier,
- WriteHandleFactory writeHandleFactory) {
+ public HoodieLazyInsertIterable(Iterator> recordItr, boolean areRecordsSorted,
+ HoodieWriteConfig config, String instantTime, HoodieTable hoodieTable,
+ String idPrefix, TaskContextSupplier taskContextSupplier,
+ WriteHandleFactory writeHandleFactory) {
super(recordItr);
this.areRecordsSorted = areRecordsSorted;
this.hoodieConfig = config;
this.instantTime = instantTime;
this.hoodieTable = hoodieTable;
this.idPrefix = idPrefix;
- this.sparkTaskContextSupplier = sparkTaskContextSupplier;
+ this.taskContextSupplier = taskContextSupplier;
this.writeHandleFactory = writeHandleFactory;
}
@@ -108,32 +104,11 @@ public class LazyInsertIterable
@Override
protected void start() {}
- @Override
- protected List computeNext() {
- // Executor service used for launching writer thread.
- BoundedInMemoryExecutor, HoodieInsertValueGenResult, List> bufferedIteratorExecutor =
- null;
- try {
- final Schema schema = new Schema.Parser().parse(hoodieConfig.getSchema());
- bufferedIteratorExecutor =
- new SparkBoundedInMemoryExecutor<>(hoodieConfig, inputItr, getInsertHandler(), getTransformFunction(schema));
- final List result = bufferedIteratorExecutor.execute();
- assert result != null && !result.isEmpty() && !bufferedIteratorExecutor.isRemaining();
- return result;
- } catch (Exception e) {
- throw new HoodieException(e);
- } finally {
- if (null != bufferedIteratorExecutor) {
- bufferedIteratorExecutor.shutdownNow();
- }
- }
- }
-
@Override
protected void end() {}
protected CopyOnWriteInsertHandler getInsertHandler() {
return new CopyOnWriteInsertHandler(hoodieConfig, instantTime, areRecordsSorted, hoodieTable, idPrefix,
- sparkTaskContextSupplier, writeHandleFactory);
+ taskContextSupplier, writeHandleFactory);
}
}
diff --git a/hudi-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertSortMode.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertSortMode.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertSortMode.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/bulkinsert/BulkInsertSortMode.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/HoodieIndex.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
similarity index 54%
rename from hudi-client/src/main/java/org/apache/hudi/index/HoodieIndex.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
index 4043586dd..c71b34e69 100644
--- a/hudi-client/src/main/java/org/apache/hudi/index/HoodieIndex.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndex.java
@@ -21,35 +21,26 @@ package org.apache.hudi.index;
import org.apache.hudi.ApiMaturityLevel;
import org.apache.hudi.PublicAPIClass;
import org.apache.hudi.PublicAPIMethod;
-import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieKey;
-import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordPayload;
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.StringUtils;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIndexException;
-import org.apache.hudi.index.bloom.HoodieBloomIndex;
-import org.apache.hudi.index.bloom.HoodieGlobalBloomIndex;
-import org.apache.hudi.index.hbase.HBaseIndex;
-import org.apache.hudi.index.simple.HoodieGlobalSimpleIndex;
-import org.apache.hudi.index.simple.HoodieSimpleIndex;
import org.apache.hudi.table.HoodieTable;
-import org.apache.spark.api.java.JavaPairRDD;
-import org.apache.spark.api.java.JavaRDD;
-import org.apache.spark.api.java.JavaSparkContext;
-
import java.io.Serializable;
/**
* Base class for different types of indexes to determine the mapping from uuid.
+ *
+ * @param Sub type of HoodieRecordPayload
+ * @param Type of inputs
+ * @param Type of keys
+ * @param Type of outputs
*/
@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
-public abstract class HoodieIndex implements Serializable {
+public abstract class HoodieIndex implements Serializable {
protected final HoodieWriteConfig config;
@@ -57,49 +48,13 @@ public abstract class HoodieIndex implements Seri
this.config = config;
}
- public static HoodieIndex createIndex(
- HoodieWriteConfig config) throws HoodieIndexException {
- // first use index class config to create index.
- if (!StringUtils.isNullOrEmpty(config.getIndexClass())) {
- Object instance = ReflectionUtils.loadClass(config.getIndexClass(), config);
- if (!(instance instanceof HoodieIndex)) {
- throw new HoodieIndexException(config.getIndexClass() + " is not a subclass of HoodieIndex");
- }
- return (HoodieIndex) instance;
- }
- switch (config.getIndexType()) {
- case HBASE:
- return new HBaseIndex<>(config);
- case INMEMORY:
- return new InMemoryHashIndex<>(config);
- case BLOOM:
- return new HoodieBloomIndex<>(config);
- case GLOBAL_BLOOM:
- return new HoodieGlobalBloomIndex<>(config);
- case SIMPLE:
- return new HoodieSimpleIndex<>(config);
- case GLOBAL_SIMPLE:
- return new HoodieGlobalSimpleIndex<>(config);
- default:
- throw new HoodieIndexException("Index type unspecified, set " + config.getIndexType());
- }
- }
-
- /**
- * Checks if the given [Keys] exists in the hoodie table and returns [Key, Option[partitionPath, fileID]] If the
- * optional is empty, then the key is not found.
- */
- @PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
- public abstract JavaPairRDD>> fetchRecordLocation(
- JavaRDD hoodieKeys, final JavaSparkContext jsc, HoodieTable hoodieTable);
-
/**
* Looks up the index and tags each incoming record with a location of a file that contains the row (if it is actually
* present).
*/
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
- public abstract JavaRDD> tagLocation(JavaRDD> recordRDD, JavaSparkContext jsc,
- HoodieTable hoodieTable) throws HoodieIndexException;
+ public abstract I tagLocation(I records, HoodieEngineContext context,
+ HoodieTable hoodieTable) throws HoodieIndexException;
/**
* Extracts the location of written records, and updates the index.
@@ -107,11 +62,11 @@ public abstract class HoodieIndex implements Seri
* TODO(vc): We may need to propagate the record as well in a WriteStatus class
*/
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
- public abstract JavaRDD updateLocation(JavaRDD writeStatusRDD, JavaSparkContext jsc,
- HoodieTable hoodieTable) throws HoodieIndexException;
+ public abstract O updateLocation(O writeStatusRDD, HoodieEngineContext context,
+ HoodieTable hoodieTable) throws HoodieIndexException;
/**
- * Rollback the efffects of the commit made at instantTime.
+ * Rollback the effects of the commit made at instantTime.
*/
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
public abstract boolean rollbackCommit(String instantTime);
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
similarity index 75%
rename from hudi-client/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
index ebd30b3ba..8cd0cb2f5 100644
--- a/hudi-client/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
@@ -18,6 +18,7 @@
package org.apache.hudi.index;
+import org.apache.hudi.client.common.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
@@ -26,8 +27,6 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.table.HoodieTable;
-import org.apache.spark.api.java.JavaSparkContext;
-
import java.util.ArrayList;
import java.util.List;
@@ -42,28 +41,26 @@ public class HoodieIndexUtils {
* Fetches Pair of partition path and {@link HoodieBaseFile}s for interested partitions.
*
* @param partitions list of partitions of interest
- * @param jsc instance of {@link JavaSparkContext} to use
+ * @param context instance of {@link HoodieEngineContext} to use
* @param hoodieTable instance of {@link HoodieTable} of interest
* @return the list of Pairs of partition path and fileId
*/
public static List> getLatestBaseFilesForAllPartitions(final List partitions,
- final JavaSparkContext jsc,
+ final HoodieEngineContext context,
final HoodieTable hoodieTable) {
- jsc.setJobGroup(HoodieIndexUtils.class.getSimpleName(), "Load latest base files from all partitions");
- return jsc.parallelize(partitions, Math.max(partitions.size(), 1))
- .flatMap(partitionPath -> {
- Option latestCommitTime = hoodieTable.getMetaClient().getCommitsTimeline()
- .filterCompletedInstants().lastInstant();
- List> filteredFiles = new ArrayList<>();
- if (latestCommitTime.isPresent()) {
- filteredFiles = hoodieTable.getBaseFileOnlyView()
- .getLatestBaseFilesBeforeOrOn(partitionPath, latestCommitTime.get().getTimestamp())
- .map(f -> Pair.of(partitionPath, f))
- .collect(toList());
- }
- return filteredFiles.iterator();
- })
- .collect();
+ context.setJobStatus(HoodieIndexUtils.class.getSimpleName(), "Load latest base files from all partitions");
+ return context.flatMap(partitions, partitionPath -> {
+ Option latestCommitTime = hoodieTable.getMetaClient().getCommitsTimeline()
+ .filterCompletedInstants().lastInstant();
+ List> filteredFiles = new ArrayList<>();
+ if (latestCommitTime.isPresent()) {
+ filteredFiles = hoodieTable.getBaseFileOnlyView()
+ .getLatestBaseFilesBeforeOrOn(partitionPath, latestCommitTime.get().getTimestamp())
+ .map(f -> Pair.of(partitionPath, f))
+ .collect(toList());
+ }
+ return filteredFiles.stream();
+ }, Math.max(partitions.size(), 1));
}
/**
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/BloomIndexFileInfo.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BloomIndexFileInfo.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/BloomIndexFileInfo.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/BloomIndexFileInfo.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/IndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IndexFileFilter.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/IndexFileFilter.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IndexFileFilter.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedGlobalIndexFileFilter.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedIndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedIndexFileFilter.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedIndexFileFilter.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/IntervalTreeBasedIndexFileFilter.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/KeyRangeLookupTree.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/KeyRangeLookupTree.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/KeyRangeLookupTree.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/KeyRangeLookupTree.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/KeyRangeNode.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/KeyRangeNode.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/KeyRangeNode.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/KeyRangeNode.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/ListBasedGlobalIndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedGlobalIndexFileFilter.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/ListBasedGlobalIndexFileFilter.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedGlobalIndexFileFilter.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/bloom/ListBasedIndexFileFilter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedIndexFileFilter.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/bloom/ListBasedIndexFileFilter.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/bloom/ListBasedIndexFileFilter.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/hbase/DefaultHBaseQPSResourceAllocator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/hbase/DefaultHBaseQPSResourceAllocator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/hbase/DefaultHBaseQPSResourceAllocator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/hbase/DefaultHBaseQPSResourceAllocator.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/index/hbase/HBaseIndexQPSResourceAllocator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/hbase/HBaseIndexQPSResourceAllocator.java
similarity index 100%
rename from hudi-client/src/main/java/org/apache/hudi/index/hbase/HBaseIndexQPSResourceAllocator.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/hbase/HBaseIndexQPSResourceAllocator.java
diff --git a/hudi-client/src/main/java/org/apache/hudi/io/AppendHandleFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/AppendHandleFactory.java
similarity index 73%
rename from hudi-client/src/main/java/org/apache/hudi/io/AppendHandleFactory.java
rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/AppendHandleFactory.java
index 4a5554bd9..5c54dce31 100644
--- a/hudi-client/src/main/java/org/apache/hudi/io/AppendHandleFactory.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/AppendHandleFactory.java
@@ -18,17 +18,17 @@
package org.apache.hudi.io;
-import org.apache.hudi.client.SparkTaskContextSupplier;
+import org.apache.hudi.client.common.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;
-public class AppendHandleFactory extends WriteHandleFactory {
+public class AppendHandleFactory extends WriteHandleFactory {
@Override
- public HoodieAppendHandle create(final HoodieWriteConfig hoodieConfig, final String commitTime,
- final HoodieTable