You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/05/10 12:00:27 UTC
[GitHub] [hudi] danny0405 commented on a diff in pull request #3599: [HUDI-2207] Support independent flink hudi clustering function
danny0405 commented on code in PR #3599:
URL: https://github.com/apache/hudi/pull/3599#discussion_r868867922
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/FlinkCustomColumnsSortPartitioner.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.sink.clustering;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.util.RowDataToAvroConverters;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.stream.Stream;
+
+/**
+ * A partitioner that does sorting based on specified column values for Java client.
+ *
+ * @param <T> HoodieRecordPayload type
+ */
Review Comment:
Unnecessary type annotation.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java:
##########
@@ -73,7 +73,7 @@ public Option<HoodieCompactionPlan> execute() {
// TODO(yihua): this validation is removed for Java client used by kafka-connect. Need to revisit this.
if (config.getEngineType() != EngineType.JAVA) {
// if there are inflight writes, their instantTime must not be less than that of compaction instant time
- table.getActiveTimeline().getCommitsTimeline().filterPendingExcludingCompaction().firstInstant()
+ table.getActiveTimeline().getCommitsTimeline().filterPendingExcludingCompactionAndReplace().firstInstant()
Review Comment:
Why this change ?
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java:
##########
@@ -96,6 +96,13 @@ public HoodieTimeline filterPendingExcludingCompaction() {
&& (!instant.getAction().equals(HoodieTimeline.COMPACTION_ACTION))), details);
}
+ @Override
+ public HoodieTimeline filterPendingExcludingCompactionAndReplace() {
+ return new HoodieDefaultTimeline(instants.stream().filter(instant -> (!instant.isCompleted())
+ && (!instant.getAction().equals(HoodieTimeline.COMPACTION_ACTION)
Review Comment:
Why this change ?
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/FlinkCustomColumnsSortPartitioner.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.sink.clustering;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.util.RowDataToAvroConverters;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.stream.Stream;
+
+/**
+ * A partitioner that does sorting based on specified column values for Java client.
+ *
+ * @param <T> HoodieRecordPayload type
+ */
+public class FlinkCustomColumnsSortPartitioner
+ implements BulkInsertPartitioner<Stream<RowData>> {
+
+ private final String[] sortColumnNames;
+ private final Schema schema;
+ private final boolean consistentLogicalTimestampEnabled;
+ private final RowDataToAvroConverters.RowDataToAvroConverter converter;
+
+ public FlinkCustomColumnsSortPartitioner(String[] columnNames, RowType rowType, Schema schema, boolean consistentLogicalTimestampEnabled) {
+ this.sortColumnNames = columnNames;
+ this.converter = RowDataToAvroConverters.createConverter(rowType);
+ this.schema = schema;
+ this.consistentLogicalTimestampEnabled = consistentLogicalTimestampEnabled;
+ }
+
+ @Override
+ public Stream<RowData> repartitionRecords(Stream<RowData> records, int outputPartitions) {
+ return records.sorted((o1, o2) -> {
+ Object values1 = HoodieAvroUtils.getRecordColumnValues(
+ (GenericRecord) converter.convert(schema, o1), sortColumnNames, consistentLogicalTimestampEnabled);
Review Comment:
You can reference the `SortOperatorGen` to get an inline sort operator for sorting. Remember to set up the transformation managed memory like https://github.com/apache/hudi/blob/6fd21d0f1043d0a06b93332d86e63d7b708fcbe8/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java#L112 did.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java:
##########
@@ -583,6 +583,66 @@ private FlinkOptions() {
.defaultValue(40)// default min 40 commits
.withDescription("Min number of commits to keep before archiving older commits into a sequential log, default 40");
+ // ------------------------------------------------------------------------
+ // Clustering Options
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<Boolean> CLUSTERING_SCHEDULE_ENABLED = ConfigOptions
+ .key("clustering.schedule.enabled")
+ .booleanType()
+ .defaultValue(false) // default false for pipeline
+ .withDescription("Schedule the cluster plan, default false");
+
+ public static final ConfigOption<Boolean> CLUSTERING_ASYNC_ENABLED = ConfigOptions
+ .key("clustering.async.enabled")
+ .booleanType()
+ .defaultValue(false) // default false for pipeline
+ .withDescription("Async Clustering, default false");
+
+ public static final ConfigOption<Integer> CLUSTERING_TASKS = ConfigOptions
+ .key("clustering.tasks")
+ .intType()
+ .defaultValue(4)
+ .withDescription("Parallelism of tasks that do actual clustering, default is 4");
+
+ public static final ConfigOption<Integer> CLUSTERING_TARGET_PARTITIONS = ConfigOptions
+ .key("clustering.plan.strategy.daybased.lookback.partitions")
+ .intType()
+ .defaultValue(2)
+ .withDescription("Number of partitions to list to create ClusteringPlan, default is 2");
+
+ public static final ConfigOption<String> CLUSTERING_PLAN_STRATEGY_CLASS = ConfigOptions
+ .key("clustering.plan.strategy.class")
+ .stringType()
+ .defaultValue("org.apache.hudi.client.clustering.plan.strategy.FlinkRecentDaysClusteringPlanStrategy")
+ .withDescription("Config to provide a strategy class (subclass of ClusteringPlanStrategy) to create clustering plan "
Review Comment:
We can use the `FlinkRecentDaysClusteringPlanStrategy .getName` to fetch the full clazz name instead.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/FlinkCustomColumnsSortPartitioner.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.sink.clustering;
+
+import org.apache.hudi.avro.HoodieAvroUtils;
+import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.util.RowDataToAvroConverters;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.stream.Stream;
+
+/**
+ * A partitioner that does sorting based on specified column values for Java client.
+ *
+ * @param <T> HoodieRecordPayload type
+ */
+public class FlinkCustomColumnsSortPartitioner
+ implements BulkInsertPartitioner<Stream<RowData>> {
+
+ private final String[] sortColumnNames;
+ private final Schema schema;
+ private final boolean consistentLogicalTimestampEnabled;
+ private final RowDataToAvroConverters.RowDataToAvroConverter converter;
+
+ public FlinkCustomColumnsSortPartitioner(String[] columnNames, RowType rowType, Schema schema, boolean consistentLogicalTimestampEnabled) {
+ this.sortColumnNames = columnNames;
+ this.converter = RowDataToAvroConverters.createConverter(rowType);
+ this.schema = schema;
+ this.consistentLogicalTimestampEnabled = consistentLogicalTimestampEnabled;
+ }
+
+ @Override
+ public Stream<RowData> repartitionRecords(Stream<RowData> records, int outputPartitions) {
+ return records.sorted((o1, o2) -> {
+ Object values1 = HoodieAvroUtils.getRecordColumnValues(
+ (GenericRecord) converter.convert(schema, o1), sortColumnNames, consistentLogicalTimestampEnabled);
Review Comment:
We better fetch the columns directly, there is no need to convert the `RowData` into avro records first. You can reference the clazz `RowDataProjection` for how to fetch the selected columns.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/BulkInsertPartitioner.java:
##########
@@ -25,20 +25,20 @@
import java.io.Serializable;
/**
- * Repartition input records into at least expected number of output spark partitions. It should give below guarantees -
- * Output spark partition will have records from only one hoodie partition. - Average records per output spark
- * partitions should be almost equal to (#inputRecords / #outputSparkPartitions) to avoid possible skews.
+ * Repartition input records into at least expected number of output partitions. It should give below guarantees -
+ * Output partition will have records from only one hoodie partition. - Average records per output
+ * partitions should be almost equal to (#inputRecords / #outputPartitions) to avoid possible skews.
*/
public interface BulkInsertPartitioner<I> extends Serializable {
/**
- * Repartitions the input records into at least expected number of output spark partitions.
+ * Repartitions the input records into at least expected number of output partitions.
*
* @param records Input Hoodie records
- * @param outputSparkPartitions Expected number of output partitions
+ * @param outputPartitions Expected number of output partitions
* @return
Review Comment:
Align the param documents.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org