-
Notifications
You must be signed in to change notification settings - Fork 1.8k
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.
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
[Feature][Connector-V2] Support multi-table sink feature for paimon #5652 #6449
[Feature][Connector-V2] Support multi-table sink feature for paimon #5652 #6449
Conversation
Please open CI on your fork repository. |
2f7ccda
to
013d68e
Compare
done |
5e3fe5e
to
7650e3f
Compare
.github/workflows/build_main.yml
Outdated
permissions: | ||
packages: write | ||
name: Run | ||
uses: ./.github/workflows/backend.yml | ||
uses: ./.github/notify_test_workflow.yml |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hi @dailai , please do not change the workflow file. It should work fine.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, it was a mistake
7650e3f
to
23dbea9
Compare
Please try merge from dev to retrigger CI. |
|
||
import java.util.concurrent.atomic.AtomicInteger; | ||
|
||
public class PaimonTypeMapper { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please extends TypeConverter
.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please extends
TypeConverter
.
ok
// Auto create if not exists the database and table for paimon | ||
catalog.createDatabase(database, true); | ||
TableSchema tableSchema = catalogTable.getTableSchema(); | ||
this.seaTunnelRowType = tableSchema.toPhysicalRowDataType(); | ||
Schema paimonTableSchema = SchemaUtil.toPaimonSchema(tableSchema); | ||
catalog.createTable(identifier, paimonTableSchema, true); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please move create table logic to SaveModeHandler
part. You can refer
seatunnel/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SupportSaveMode.java
Line 23 in 6ec16ac
public interface SupportSaveMode { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please move create table logic to
SaveModeHandler
part. You can referseatunnel/seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SupportSaveMode.java
Line 23 in 6ec16ac
public interface SupportSaveMode {
done, please review!
<dependency> | ||
<groupId>org.xerial.snappy</groupId> | ||
<artifactId>snappy-java</artifactId> | ||
<version>1.1.10.4</version> | ||
</dependency> |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Any reason for add this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we shoud not put seatunnel-hadoop3-3.1.4-uber
into flink libs. It only used for zeta. With flink, we should use their own hadoop jar.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we shoud not put
seatunnel-hadoop3-3.1.4-uber
into flink libs. It only used for zeta. With flink, we should use their own hadoop jar.
I did not put the seatunnel-hadoop3-3.1.4-uber into flink libs. If we submit a job by the seatunnel's flink script , the flink will load the seatunnel-hadoop3-3.1.4-uber in classpath. For example, exec "start-seatunnel-flink-15-connector-v2.sh --config xxxx.job"(The source is mysql-cdc and sink is paimon )
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@Hisoka-X please recheck
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for ping me. I will check it this weekend.
23dbea9
to
d1925e8
Compare
2e0943b
to
6d44826
Compare
[Feature][Paimon] Support cdc write of paimon sink apache#6427 [Feature][Paimon] Support auto create db and table for paimon sink
a003e52
to
cd7c21a
Compare
cd7c21a
to
5b32488
Compare
} | ||
|
||
source { | ||
MySQL-CDC { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You can use FakeSource
instead of MySQL-CDC
, the test case is more stable, please refer to https://github.com/apache/seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/clickhouse_sink_cdc_changelog_case2.conf#L27C3-L27C13
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please change the docs |
|
317712e
to
bf918e4
Compare
@@ -0,0 +1,65 @@ | |||
# |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
remove this file?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done
@@ -0,0 +1,24 @@ | |||
-- |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
remove this file?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
cc @TyrantLucifer
import static org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonConfig.HDFS_SITE_PATH; | ||
import static org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonConfig.TABLE; | ||
import static org.apache.seatunnel.connectors.seatunnel.paimon.config.PaimonConfig.WAREHOUSE; | ||
import static org.apache.seatunnel.api.table.factory.FactoryUtil.discoverFactory; | ||
|
||
@AutoService(SeaTunnelSink.class) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to remove this annotation?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to remove this annotation?
I have removed it.
<dependencies> | ||
<dependency> | ||
<groupId>org.apache.seatunnel</groupId> | ||
<artifactId>connector-jdbc</artifactId> |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can this dependency be removed?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done
"sh", "-c", "cd " + CATALOG_ROOT_DIR + " && tar -zxvf " + NAMESPACE_TAR); | ||
try { | ||
Process process = processBuilder.start(); | ||
// 等待命令执行完成 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please use English comments.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done
- [x] [exactly-once](../../concept/connector-v2-features.md) | ||
|
||
## 连接器选项 | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There seem to be some problems with the format
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
8a1d736
to
b5474a9
Compare
b5474a9
to
408784f
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
Purpose of this pull request
[Feature][Paimon] Support cdc write of paimon sink #6427
[Feature][Paimon] Support the streaming mode for sink #6342
[Feature][Paimon] Support save mode
[Feature][Paimon] Support multi-table sink feature #5652
Does this PR introduce any user-facing change?
no
How was this patch tested?
add new test.
Check list
New License Guide
release-note
.