datastream sink support dynamic bucketing

Signed-off-by: zenghua <huazeng@dmetasoul.com>
lakesoul-io · Ceng23333 · Jul 26, 2024 · Jul 26, 2024 · Jul 30, 2024 · Jul 30, 2024
commit 4cac81643a2e4bb9eaae48567be207eb2e62301f
diff --git a/.github/workflows/flink-cdc-hdfs-test.yml b/.github/workflows/flink-cdc-hdfs-test.yml
@@ -122,15 +122,15 @@ jobs:
           sleep 30s
       - name: Start flink mysql cdc task-1
         run: |
-          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.entry.MysqlCdc /opt/flink/work-dir/$FLINK_JAR_NAME --source_db.host mysql --source_db.port 3306 --source_db.db_name test_cdc --source_db.user root --source_db.password root --source.parallelism 2 --sink.parallelism 4 --use.cdc true --warehouse_path hdfs://172.17.0.1:9000/lakesoul-test-bucket/data/ --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/chk --flink.savepoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/svp --job.checkpoint_interval 30000 --server_time_zone UTC
+          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.entry.MysqlCdc /opt/flink/work-dir/$FLINK_JAR_NAME --source_db.host mysql --source_db.port 3306 --source_db.db_name test_cdc --source_db.user root --source_db.password root --source.parallelism 2 --sink.parallelism 4 --use.cdc true --warehouse_path hdfs://172.17.0.1:9000/lakesoul-test-bucket/data/ --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/chk --flink.savepoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/svp --job.checkpoint_interval 5000 --server_time_zone UTC
           sleep 30s
       - name: Start flink source to sink task-2
         run: |
-          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.test.benchmark.LakeSoulSourceToSinkTable -C file:///opt/flink/work-dir/$FLINK_JAR_NAME /opt/flink/work-dir/$FLINK_TEST_JAR_NAME --source.database.name test_cdc --source.table.name default_init --sink.database.name flink_sink --sink.table.name default_init --use.cdc true --hash.bucket.number 2 --job.checkpoint_interval 30000 --server_time_zone UTC --warehouse.path hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink-sink/data --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink-sink/chk
+          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.test.benchmark.LakeSoulSourceToSinkTable -C file:///opt/flink/work-dir/$FLINK_JAR_NAME /opt/flink/work-dir/$FLINK_TEST_JAR_NAME --source.database.name test_cdc --source.table.name default_init --sink.database.name flink_sink --sink.table.name default_init --use.cdc true --hash.bucket.number 2 --job.checkpoint_interval 10000 --server_time_zone UTC --warehouse.path hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink-sink/data --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink-sink/chk
           sleep 30s
       - name: Start flink DataGenSource without primary key task-3
         run: |
-          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.test.benchmark.LakeSoulDataGenSourceTable -C file:///opt/flink/work-dir/$FLINK_JAR_NAME /opt/flink/work-dir/$FLINK_TEST_JAR_NAME --sink.database.name flink --sink.table.name sink_table --job.checkpoint_interval 30000 --server_time_zone UTC --warehouse.path hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink/ --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink/chk --sink.parallel 2 --data.size 1000 --write.time 5
+          docker exec -t -u flink lakesoul-docker-compose-env-jobmanager-1 flink run -d -c org.apache.flink.lakesoul.test.benchmark.LakeSoulDataGenSourceTable -C file:///opt/flink/work-dir/$FLINK_JAR_NAME /opt/flink/work-dir/$FLINK_TEST_JAR_NAME --sink.database.name flink --sink.table.name sink_table --job.checkpoint_interval 10000 --server_time_zone UTC --warehouse.path hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink/ --flink.checkpoint hdfs://172.17.0.1:9000/lakesoul-test-bucket/flink/chk --sink.parallel 2 --data.size 1000 --write.time 5
       - name: Download mysql driver jar
         run: |
           cd ./script/benchmark/work-dir

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/JdbcCDC.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/JdbcCDC.java
@@ -65,7 +65,7 @@ public static void main(String[] args) throws Exception {
         host = parameter.get(SOURCE_DB_HOST.key());
         port = parameter.getInt(SOURCE_DB_PORT.key(), MysqlDBManager.DEFAULT_MYSQL_PORT);
         //Postgres Oracle
-        if (dbType.equalsIgnoreCase("oracle") || dbType.equalsIgnoreCase("postgres") ) {
+        if (dbType.equalsIgnoreCase("oracle") || dbType.equalsIgnoreCase("postgres")) {
             schemaList = parameter.get(SOURCE_DB_SCHEMA_LIST.key()).split(",");
             String[] tables = parameter.get(SOURCE_DB_SCHEMA_TABLES.key()).split(",");
             tableList = new String[tables.length];
@@ -74,10 +74,10 @@ public static void main(String[] args) throws Exception {
             }
             splitSize = parameter.getInt(SOURCE_DB_SPLIT_SIZE.key(), SOURCE_DB_SPLIT_SIZE.defaultValue());
         }
-        if (dbType.equalsIgnoreCase("sqlserver") ){
+        if (dbType.equalsIgnoreCase("sqlserver")) {
             tableList = parameter.get(SOURCE_DB_SCHEMA_TABLES.key()).split(",");
         }
-        if ( dbType.equalsIgnoreCase("mongodb")){
+        if (dbType.equalsIgnoreCase("mongodb")) {
             batchSize = parameter.getInt(BATCH_SIZE.key(), BATCH_SIZE.defaultValue());
             tableList = parameter.get(SOURCE_DB_SCHEMA_TABLES.key()).split(",");
         }
@@ -99,11 +99,11 @@ public static void main(String[] args) throws Exception {
         conf.set(SOURCE_DB_PORT, port);
         conf.set(WAREHOUSE_PATH, databasePrefixPath);
         conf.set(SERVER_TIME_ZONE, serverTimezone);
-        conf.set(SOURCE_DB_TYPE,dbType);
+        conf.set(SOURCE_DB_TYPE, dbType);
 
         // parameters for mutil tables dml sink
         conf.set(LakeSoulSinkOptions.USE_CDC, true);
-        conf.set(LakeSoulSinkOptions.isMultiTableSource, true);
+        conf.set(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE, true);
         conf.set(LakeSoulSinkOptions.WAREHOUSE_PATH, databasePrefixPath);
         conf.set(LakeSoulSinkOptions.SOURCE_PARALLELISM, sourceParallelism);
         conf.set(LakeSoulSinkOptions.BUCKET_PARALLELISM, bucketParallelism);

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/MysqlCdc.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/MysqlCdc.java
@@ -74,9 +74,9 @@ public static void main(String[] args) throws Exception {
         conf.set(WAREHOUSE_PATH, databasePrefixPath);
         conf.set(SERVER_TIME_ZONE, serverTimezone);
 
-        // parameters for mutil tables dml sink
+        // parameters for multi tables dml sink
         conf.set(LakeSoulSinkOptions.USE_CDC, true);
-        conf.set(LakeSoulSinkOptions.isMultiTableSource, true);
+        conf.set(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE, true);
         conf.set(LakeSoulSinkOptions.WAREHOUSE_PATH, databasePrefixPath);
         conf.set(LakeSoulSinkOptions.SOURCE_PARALLELISM, sourceParallelism);
         conf.set(LakeSoulSinkOptions.BUCKET_PARALLELISM, bucketParallelism);

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/OracleCdc.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/OracleCdc.java
@@ -73,7 +73,7 @@ public static void main(String[] args) throws Exception {
         }
         Configuration conf = new Configuration();
         conf.set(LakeSoulSinkOptions.USE_CDC, true);
-        conf.set(LakeSoulSinkOptions.isMultiTableSource, true);
+        conf.set(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE, true);
         conf.set(SOURCE_PARALLELISM, sourceParallelism);
         conf.set(BUCKET_PARALLELISM, bucketParallelism);
         conf.set(SERVER_TIME_ZONE, serverTimezone);

diff --git a/...main/java/org/apache/flink/lakesoul/sink/writer/AbstractLakeSoulMultiTableSinkWriter.java b/...main/java/org/apache/flink/lakesoul/sink/writer/AbstractLakeSoulMultiTableSinkWriter.java
@@ -147,6 +147,7 @@ protected long getDataDmlTsMs(IN element) {
 
     @Override
     public void write(IN element, Context context) throws IOException {
+        LOG.info("{}", element);
         if (element == null) {
             return;
         }
@@ -163,6 +164,7 @@ public void write(IN element, Context context) throws IOException {
         } catch (Exception e) {
             throw new IOException(e);
         }
+        LOG.info("{}", schemaAndRowDatas);
         for (Tuple2<TableSchemaIdentity, RowData> schemaAndRowData : schemaAndRowDatas) {
             TableSchemaIdentity identity = schemaAndRowData.f0;
             RowData rowData = schemaAndRowData.f1;

diff --git a/.../main/java/org/apache/flink/lakesoul/sink/writer/DynamicPartitionNativeParquetWriter.java b/.../main/java/org/apache/flink/lakesoul/sink/writer/DynamicPartitionNativeParquetWriter.java
@@ -77,10 +77,11 @@ public DynamicPartitionNativeParquetWriter(RowType rowType,
     private void initNativeWriter() throws IOException {
         ArrowUtils.setLocalTimeZone(FlinkUtil.getLocalTimeZone(conf));
         Schema arrowSchema = ArrowUtils.toArrowSchema(rowType);
+        System.out.println(arrowSchema);
         nativeWriter = new NativeIOWriter(arrowSchema);
         nativeWriter.setPrimaryKeys(primaryKeys);
         nativeWriter.setRangePartitions(rangeColumns);
-        if (conf.getBoolean(LakeSoulSinkOptions.isMultiTableSource)) {
+        if (conf.getBoolean(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE)) {
             nativeWriter.setAuxSortColumns(Collections.singletonList(SORT_FIELD));
         }
         nativeWriter.setHashBucketNum(conf.getInteger(LakeSoulSinkOptions.HASH_BUCKET_NUM));
@@ -89,7 +90,7 @@ private void initNativeWriter() throws IOException {
         batch = VectorSchemaRoot.create(arrowSchema, nativeWriter.getAllocator());
         arrowWriter = ArrowUtils.createRowDataArrowWriter(batch, rowType);
 
-
+        System.out.println(prefix);
         nativeWriter.withPrefix(this.prefix);
         nativeWriter.useDynamicPartition(true);
 

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/writer/NativeParquetWriter.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/sink/writer/NativeParquetWriter.java
@@ -71,7 +71,7 @@ public NativeParquetWriter(RowType rowType,
         Schema arrowSchema = ArrowUtils.toArrowSchema(rowType);
         nativeWriter = new NativeIOWriter(arrowSchema);
         nativeWriter.setPrimaryKeys(primaryKeys);
-        if (conf.getBoolean(LakeSoulSinkOptions.isMultiTableSource)) {
+        if (conf.getBoolean(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE)) {
             nativeWriter.setAuxSortColumns(Collections.singletonList(SORT_FIELD));
         }
         nativeWriter.setRowGroupRowNumber(this.maxRowGroupRows);

diff --git a/...in/java/org/apache/flink/lakesoul/sink/writer/arrow/NativeLakeSoulArrowWrapperWriter.java b/...in/java/org/apache/flink/lakesoul/sink/writer/arrow/NativeLakeSoulArrowWrapperWriter.java
@@ -75,7 +75,7 @@ private void initNativeWriter() throws IOException {
         nativeWriter = new NativeIOWriter(arrowSchema);
         nativeWriter.setPrimaryKeys(primaryKeys);
         nativeWriter.setRangePartitions(rangeColumns);
-        if (conf.getBoolean(LakeSoulSinkOptions.isMultiTableSource)) {
+        if (conf.getBoolean(LakeSoulSinkOptions.IS_MULTI_TABLE_SOURCE)) {
             nativeWriter.setAuxSortColumns(Collections.singletonList(SORT_FIELD));
         }
         nativeWriter.setHashBucketNum(conf.getInteger(LakeSoulSinkOptions.HASH_BUCKET_NUM));

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/tool/LakeSoulSinkOptions.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/tool/LakeSoulSinkOptions.java
@@ -32,6 +32,7 @@ public class LakeSoulSinkOptions {
 
     public static final String CDC_CHANGE_COLUMN_DEFAULT = "rowKinds";
 
+    @Deprecated
     public static final String SORT_FIELD = "__sort_filed__";
 
     public static final Long DEFAULT_BUCKET_ROLLING_SIZE = 1000000L;
@@ -110,7 +111,8 @@ public class LakeSoulSinkOptions {
             .booleanType()
             .defaultValue(false)
             .withDescription("use cdc column ");
-    public static final ConfigOption<Boolean> isMultiTableSource = ConfigOptions
+    @Deprecated
+    public static final ConfigOption<Boolean> IS_MULTI_TABLE_SOURCE = ConfigOptions
             .key("Multi_Table_Source")
             .booleanType()
             .defaultValue(false)

diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecord.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecord.java
@@ -87,7 +87,7 @@ public static BinarySourceRecord fromMysqlSourceRecord(SourceRecord sourceRecord
             Struct source = value.getStruct(Envelope.FieldName.SOURCE);
             if (sourceField != null && source != null) {
                 if (sourceField.schema().field("file") != null) {
-                    String fileName = (String)source.getWithoutDefault("file");
+                    String fileName = (String) source.getWithoutDefault("file");
                     if (StringUtils.isNotBlank(fileName)) {
                         binlogFileIndex = Long.parseLong(fileName.substring(fileName.lastIndexOf(".") + 1));
                     }
@@ -99,12 +99,13 @@ public static BinarySourceRecord fromMysqlSourceRecord(SourceRecord sourceRecord
                     tsMs = (Long) source.getWithoutDefault("ts_ms");
                 }
             }
+            // sortField has been deprecated
             long sortField = (binlogFileIndex << 32) + binlogPosition;
             LakeSoulRowDataWrapper data = convert.toLakeSoulDataType(valueSchema, value, tableId, tsMs, sortField);
             String tablePath;
-            if (tableId.schema()==null){
+            if (tableId.schema() == null) {
                 tablePath = new Path(new Path(basePath, tableId.catalog()), tableId.table()).toString();
-            }else {
+            } else {
                 tablePath = new Path(new Path(basePath, tableId.schema()), tableId.table()).toString();
             }
             return new BinarySourceRecord(sourceRecord.topic(), primaryKeys, tableId, FlinkUtil.makeQualifiedPath(tablePath).toString(),