捣鼓2天拉 ,为啥 doris cdc 就是没同步成功数据完全按照文档来的呀有大佬吗 #11453

ChewJ404 · 2022-08-03T02:57:58Z

ChewJ404
Aug 3, 2022

配置环境 The environment

flink:1.14.5-scala_2.12-java8
Doris 1.1
flink-doris-connector-1.14_2.12/1.1.0

操作步骤

doris 建表 doris create table
源表

CREATE TABLE `user` (
    `id` bigint(20) NOT NULL COMMENT "用户id",
    `name` varchar(50) NOT NULL COMMENT "用户昵称"
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
  "replication_allocation" = "tag.location.default: 1",
  "in_memory" = "false",
  "storage_format" = "V2"
);

sink表

CREATE TABLE `user_sink` (
    `id` bigint(20) NOT NULL COMMENT "用户id",
    `name` varchar(50) NOT NULL COMMENT "用户昵称"
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
  "replication_allocation" = "tag.location.default: 1",
  "in_memory" = "false",
  "storage_format" = "V2"
)

源表插入数据

INSERT INTO `dataease`.`user` (`id`, `name`) VALUES (1, 'aaaa');
INSERT INTO `dataease`.`user` (`id`, `name`) VALUES (2, 'bbbb');

创建 flink-sql 任务 crate flink-sql task
创建 resource

CREATE TABLE flink_doris_source (
    `id` BIGINT,
    `name` STRING
) WITH (
    'connector' = 'doris',
    'fenodes' = '192.168.12.222:8030',
    'table.identifier' = 'dataease.user',
    'username' = 'root',
    'password' = '123456'
);

在flink-sql 中查询能查询到源表数据 2条'
select * from flink_doris_source;
创建 sink

CREATE TABLE flink_doris_sink (
    `id` BIGINT,
    `name` STRING
) WITH (
    'connector' = 'doris',
    'fenodes' = '192.168.12.222:8030',
    'table.identifier' = 'dataease.user_sink',
    'username' = 'root',
    'password' = '123456',
    'sink.label-prefix' = 'doris_label'
);

执行同步 submit job
INSERT INTO flink_doris_sink select id,name from flink_doris_source;

结果

数据没有同步成功查看 user_sink 内还是空的 ,flink job 里面也没有报错 ,想知道如何排查问题呢

Answered by hf200012

Aug 5, 2022

SET execution.checkpointing.interval = 10s;

View full answer

ChewJ404 · 2022-08-03T03:03:27Z

ChewJ404
Aug 3, 2022
Author

调整了还是没数据

2022-08-03 04:00:59,036 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Receive slot request 4f19c9acad182240e234de3ffbb05215 for job 35cb94c0780d0fbfedb6e140be479fca from resource manager with leader id 00000000000000000000000000000000.
2022-08-03 04:00:59,041 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Allocated slot for 4f19c9acad182240e234de3ffbb05215.
2022-08-03 04:00:59,041 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Add job 35cb94c0780d0fbfedb6e140be479fca for job leader monitoring.
2022-08-03 04:00:59,042 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Try to register at job manager akka.tcp://flink@jobmanager:6123/user/rpc/jobmanager_4 with leader id 00000000-0000-0000-0000-000000000000.
2022-08-03 04:00:59,060 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Resolved JobManager address, beginning registration
2022-08-03 04:00:59,075 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Successful registration at job manager akka.tcp://flink@jobmanager:6123/user/rpc/jobmanager_4 for job 35cb94c0780d0fbfedb6e140be479fca.
2022-08-03 04:00:59,075 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Establish JobManager connection for job 35cb94c0780d0fbfedb6e140be479fca.
2022-08-03 04:00:59,076 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Offer reserved slots to the leader of job 35cb94c0780d0fbfedb6e140be479fca.
2022-08-03 04:00:59,087 INFO  org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl [] - Activate slot 4f19c9acad182240e234de3ffbb05215.
2022-08-03 04:00:59,092 INFO  org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader [] - Creating a changelog storage with name 'memory'.
2022-08-03 04:00:59,094 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Received task Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d), deploy into slot with allocation id 4f19c9acad182240e234de3ffbb05215.
2022-08-03 04:00:59,095 INFO  org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl [] - Activate slot 4f19c9acad182240e234de3ffbb05215.
2022-08-03 04:00:59,095 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d) switched from CREATED to DEPLOYING.
2022-08-03 04:00:59,101 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Loading JAR files for task Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d) [DEPLOYING].
2022-08-03 04:00:59,113 INFO  org.apache.flink.runtime.blob.BlobClient                     [] - Downloading 35cb94c0780d0fbfedb6e140be479fca/p-e9435fe7360ee6d1ec55633df93ff2ad28e2e3d8-003eb988521fd98326f5fc13c7dd1d9b from jobmanager/172.19.0.2:6124
2022-08-03 04:00:59,131 INFO  org.apache.flink.runtime.blob.BlobClient                     [] - Downloading 35cb94c0780d0fbfedb6e140be479fca/p-2644d6d3c055fdc8327bfeadfac58623ae6163f5-498ca1d16929e516eef66ac1f9d05523 from jobmanager/172.19.0.2:6124
2022-08-03 04:00:59,322 INFO  org.apache.flink.streaming.runtime.tasks.StreamTask          [] - No state backend has been configured, using default (HashMap) org.apache.flink.runtime.state.hashmap.HashMapStateBackend@6b02323c
2022-08-03 04:00:59,322 INFO  org.apache.flink.runtime.state.StateBackendLoader            [] - State backend loader loads the state backend as HashMapStateBackend
2022-08-03 04:00:59,322 INFO  org.apache.flink.streaming.runtime.tasks.StreamTask          [] - Checkpoint storage is set to 'jobmanager'
2022-08-03 04:00:59,323 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d) switched from DEPLOYING to INITIALIZING.
2022-08-03 04:00:59,343 WARN  org.apache.flink.metrics.MetricGroup                         [] - The operator name Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) exceeded the 80 characters length limit and was truncated.
2022-08-03 04:00:59,350 WARN  org.apache.flink.metrics.MetricGroup                         [] - The operator name Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) exceeded the 80 characters length limit and was truncated.
2022-08-03 04:00:59,356 INFO  org.apache.doris.flink.sink.writer.DorisWriter               [] - restore checkpointId 0
2022-08-03 04:00:59,357 INFO  org.apache.doris.flink.sink.writer.DorisWriter               [] - labelPrefix doris_label2
2022-08-03 04:00:59,359 INFO  org.apache.doris.flink.sink.writer.DorisWriter               [] - Send request to Doris FE 'http://192.168.12.222:8030/api/backends?is_alive=true' with user 'root'.
2022-08-03 04:00:59,375 INFO  org.apache.doris.flink.sink.writer.DorisWriter               [] - Backend Info:{"backends":[{"ip":"192.168.12.222","http_port":8040,"is_alive":true}]}
2022-08-03 04:00:59,381 INFO  org.apache.doris.flink.sink.writer.RecordBuffer              [] - init RecordBuffer capacity 262144, count 3
2022-08-03 04:00:59,385 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - abort for labelSuffix doris_label2_0. start chkId 1.
2022-08-03 04:00:59,406 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - load Result {
    "TxnId": 40,
    "Label": "doris_label2_0_1",
    "TwoPhaseCommit": "true",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 0,
    "NumberLoadedRows": 0,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 0,
    "LoadTimeMs": 8,
    "BeginTxnTimeMs": 0,
    "StreamLoadPutTimeMs": 1,
    "ReadDataTimeMs": 0,
    "WriteDataTimeMs": 2,
    "CommitAndPublishTimeMs": 0
}

2022-08-03 04:00:59,406 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - abort 40 for check label doris_label2_0_1.
2022-08-03 04:00:59,531 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - abort for labelSuffix doris_label2_0 finished
2022-08-03 04:00:59,531 INFO  org.apache.doris.flink.sink.writer.RecordBuffer              [] - start buffer data, read queue size 0, write queue size 3
2022-08-03 04:00:59,531 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - stream load started for doris_label2_0_1
2022-08-03 04:00:59,532 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - start execute load
2022-08-03 04:00:59,535 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d) switched from INITIALIZING to RUNNING.
2022-08-03 04:00:59,541 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Adding split(s) to reader: [DorisSourceSplit: dataease.user,be=192.168.12.222:9060,tablets=[10048]]
2022-08-03 04:00:59,542 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Starting split fetcher 0
2022-08-03 04:00:59,597 INFO  org.apache.doris.flink.backend.BackendClient                 [] - CloseScanner to Doris BE 'Doris BE{host='192.168.12.222', port=9060}' success.
2022-08-03 04:00:59,597 INFO  org.apache.doris.flink.backend.BackendClient                 [] - Closed a connection to Doris BE{host='192.168.12.222', port=9060}.
2022-08-03 04:00:59,597 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Finished reading from splits [192.168.12.222:9060]
2022-08-03 04:00:59,597 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Finished reading split(s) [192.168.12.222:9060]
2022-08-03 04:00:59,598 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager [] - Closing splitFetcher 0 because it is idle.
2022-08-03 04:00:59,598 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Shutting down split fetcher 0
2022-08-03 04:00:59,598 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Split fetcher 0 exited.
2022-08-03 04:00:59,602 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Reader received NoMoreSplits event.
2022-08-03 04:00:59,603 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - stream load stopped.
2022-08-03 04:00:59,721 INFO  org.apache.doris.flink.sink.writer.DorisStreamLoad           [] - load Result {
    "TxnId": 41,
    "Label": "doris_label2_0_1",
    "TwoPhaseCommit": "true",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 2,
    "NumberLoadedRows": 2,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 16,
    "LoadTimeMs": 182,
    "BeginTxnTimeMs": 0,
    "StreamLoadPutTimeMs": 1,
    "ReadDataTimeMs": 0,
    "WriteDataTimeMs": 71,
    "CommitAndPublishTimeMs": 0
}

2022-08-03 04:00:59,723 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Closing Source Reader.
2022-08-03 04:00:59,724 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d) switched from RUNNING to FINISHED.
2022-08-03 04:00:59,724 INFO  org.apache.flink.runtime.taskmanager.Task                    [] - Freeing task resources for Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 (7737ffd503218c491afb14963167f59d).
2022-08-03 04:00:59,725 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Un-registering task and sending final execution state FINISHED to JobManager for task Source: TableSourceScan(table=[[default_catalog, default_database, flink_doris_source]], fields=[id, name]) -> Sink Sink(table=[default_catalog.default_database.flink_doris_sink], fields=[id, name]) (1/1)#0 7737ffd503218c491afb14963167f59d.
2022-08-03 04:00:59,748 INFO  org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl [] - Free slot TaskSlot(index:2, state:ACTIVE, resource profile: ResourceProfile{cpuCores=1, taskHeapMemory=778.240mb (816043776 bytes), taskOffHeapMemory=0 bytes, managedMemory=604.160mb (633507685 bytes), networkMemory=102.400mb (107374182 bytes)}, allocationId: 4f19c9acad182240e234de3ffbb05215, jobId: 35cb94c0780d0fbfedb6e140be479fca).
2022-08-03 04:00:59,749 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Remove job 35cb94c0780d0fbfedb6e140be479fca from job leader monitoring.
2022-08-03 04:00:59,749 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Close JobManager connection for job 35cb94c0780d0fbfedb6e140be479fca.

0 replies

JNSimba · 2022-08-03T10:22:47Z

JNSimba
Aug 3, 2022
Collaborator

Checkpoint is not enabled? You can refer to here https://mp.weixin.qq.com/s/ejDChH5ScUK4pSxo2EH_4w

1 reply

bb33chen Mar 12, 2026

Does the flink-doris-connector support scenarios where both the upstream and downstream are Doris?
（flink-doris-connector 支持上下游都是Doris的情况吗？）

keyuquan · 2022-08-03T11:21:03Z

keyuquan
Aug 3, 2022

要先设置 checkpoint，否则doris sink 不执行： SET 'execution.checkpointing.interval' = '10s';

0 replies

hf200012 · 2022-08-05T08:08:37Z

hf200012
Aug 5, 2022
Collaborator

SET execution.checkpointing.interval = 10s;

1 reply

bb33chen Mar 12, 2026

Does the flink-doris-connector support scenarios where both the upstream and downstream are Doris?
（flink-doris-connector 支持上下游都是Doris的情况吗？）

lu314401633 · 2023-05-26T07:49:35Z

lu314401633
May 26, 2023

没设置checkpoint interval，导致你执行完sql，任务就结束了

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

捣鼓2天拉 ,为啥 doris cdc 就是没同步成功数据完全按照文档来的呀有大佬吗 #11453

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 5 comments 2 replies

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

捣鼓2天拉 ,为啥 doris cdc 就是 没同步成功数据 完全按照 文档来的呀 有大佬吗 #11453

Uh oh!

Uh oh!

ChewJ404 Aug 3, 2022

配置 环境 The environment

操作步骤

结果

Replies: 5 comments · 2 replies

Uh oh!

Uh oh!

ChewJ404 Aug 3, 2022 Author

Uh oh!

JNSimba Aug 3, 2022 Collaborator

Uh oh!

bb33chen Mar 12, 2026

Uh oh!

keyuquan Aug 3, 2022

Uh oh!

hf200012 Aug 5, 2022 Collaborator

Uh oh!

bb33chen Mar 12, 2026

Uh oh!

lu314401633 May 26, 2023

捣鼓2天拉 ,为啥 doris cdc 就是没同步成功数据完全按照文档来的呀有大佬吗 #11453

ChewJ404
Aug 3, 2022

配置环境 The environment

Replies: 5 comments 2 replies

ChewJ404
Aug 3, 2022
Author

JNSimba
Aug 3, 2022
Collaborator

keyuquan
Aug 3, 2022

hf200012
Aug 5, 2022
Collaborator

lu314401633
May 26, 2023