Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

[Flink]Datastream sink supports dynamic bucketing #520

Open
wants to merge 21 commits into
base: main
Choose a base branch
from
Open

[Flink]Datastream sink supports dynamic bucketing #520

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
46181fe
parse csair log
Jul 26, 2024
6622b11
enlarge ci checkpoint interval
Jul 26, 2024
4cac816
datastream sink support dynamic bucketing
Jul 30, 2024
871013d
remove redundant log
Jul 30, 2024
cd41b1d
fix clippy
Jul 30, 2024
56185b0
update verify result format
Jul 30, 2024
c71d1f1
update verify result format
Jul 30, 2024
6b94ceb
fix case of schema migration
Jul 31, 2024
6347ae0
fix case of schema migration
Jul 31, 2024
8ab7caf
run ci with DYNAMIC_BUCKETING=false
Jul 31, 2024
5fbdf95
update ci log info
Jul 31, 2024
096bac8
merge LakeSoulMultiTableSinkGlobalCommittable with sorted createtime
Jul 31, 2024
7b9c9ee
merge LakeSoulMultiTableSinkGlobalCommittable with sorted createtime
Jul 31, 2024
b885474
fix schema migration
Jul 31, 2024
fe1ca65
fix schema migration
Jul 31, 2024
4235e04
use debezium ts_ms as latestUpdateTime
Aug 1, 2024
26e43f8
update schema with record of largest debezium ts_ms
Aug 1, 2024
ff9b3c1
update schema with record of largest debezium ts_ms
Aug 1, 2024
81024d0
add debug info
Aug 1, 2024
9b64b0c
add debug info
Aug 1, 2024
0e63939
test dynamic_bucket failover recover
Aug 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions lakesoul-common/src/main/scala/com/dmetasoul/lakesoul/meta/DataOperation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,6 +137,7 @@ object DataOperation {
metaPartitionInfoScala.setPartitionDesc(partition_info.range_value)
metaPartitionInfoScala.addAllSnapshot(JavaConverters.bufferAsJavaList(partition_info.read_files.map(DBUtil.toProtoUuid).toBuffer))
val dataCommitInfoList = dbManager.getTableSinglePartitionDataInfo(metaPartitionInfoScala.build).asScala.toArray
println(dataCommitInfoList.mkString("Array(", ", ", ")"))
for (metaDataCommitInfo <- dataCommitInfoList) {
val fileOps = metaDataCommitInfo.getFileOpsList.asScala.toArray
for (file <- fileOps) {
Expand Down
10 changes: 5 additions & 5 deletions lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/JdbcCDC.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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];
Expand All @@ -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(",");
}
Expand All @@ -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);
Expand Down
4 changes: 2 additions & 2 deletions lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/MysqlCdc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
2 changes: 1 addition & 1 deletion lakesoul-flink/src/main/java/org/apache/flink/lakesoul/entry/OracleCdc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
2 changes: 1 addition & 1 deletion ...ink/src/main/java/org/apache/flink/lakesoul/sink/LakeSoulMultiTableSinkStreamBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ public DataStream<BinarySourceRecord> buildHashPartitionedCDCStream(DataStream<B
}

public DataStreamSink<BinarySourceRecord> buildLakeSoulDMLSink(DataStream<BinarySourceRecord> stream) {
context.conf.set(DYNAMIC_BUCKETING, false);
context.conf.set(DYNAMIC_BUCKETING, true);
LakeSoulRollingPolicyImpl<RowData> rollingPolicy = new LakeSoulRollingPolicyImpl<>(
context.conf.getLong(FILE_ROLLING_SIZE), context.conf.getLong(FILE_ROLLING_TIME));
OutputFileConfig fileNameConfig = OutputFileConfig.builder()
Expand Down
2 changes: 1 addition & 1 deletion ...l-flink/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,8 @@ public List<LakeSoulMultiTableSinkCommittable> commit(List<LakeSoulMultiTableSin
for (String file : files) {
DataFileOp.Builder dataFileOp = DataFileOp.newBuilder();
dataFileOp.setFileOp(FileOp.add);
dataFileOp.setPath(file);
Path path = new Path(file);
dataFileOp.setPath(path.toString());
FileStatus fileStatus = FileSystem.get(path.toUri()).getFileStatus(path);
dataFileOp.setSize(fileStatus.getLen());
dataFileOp.setFileExistCols(fileExistCols);
Expand Down
Loading
Loading