-
|
例如如下SQL，将MySQL的变更用canal-json格式，输出到Kafka中。 CREATE TABLE orders (
order_id INT,
order_date TIMESTAMP(0),
customer_name STRING,
price DECIMAL(10, 5),
product_id INT,
order_status BOOLEAN,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '3306',
'username' = 'root',
'password' = '123456',
'database-name' = 'mydb',
'table-name' = 'orders'
);
CREATE TABLE kafka_canal_orders (
order_id INT,
order_date TIMESTAMP(0),
customer_name STRING,
price DECIMAL(10, 5),
product_id INT,
order_status BOOLEAN,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'kafka_canal_orders',
'properties.bootstrap.servers' = 'localhost:9092',
'scan.startup.mode' = 'earliest-offset',
'format' = 'canal-json'
);
INSERT INTO kafka_canal_orders SELECT * FROM orders;写入Kafka的数据，如以下格式： {"data":[{"order_id":10001,"order_date":"2020-07-30 10:08:22","customer_name":"Jark","price":50.5,"product_id":102,"order_status":false}],"type":"INSERT"}这里面丢失了 变成如下格式： {"database":"mydb", "table":"orders","ts":1636967147000,"data":[{"order_id":10001,"order_date":"2020-07-30 10:08:22","customer_name":"Jark","price":50.5,"product_id":102,"order_status":false}],"type":"INSERT"} |
Beta Was this translation helpful? Give feedback.
Answered by
ashulin
Nov 15, 2021
Replies: 1 comment 13 replies
-
|
2.1支持元数据列（即表名库名等信息）；但是序列化出来不是这个格式； |
Beta Was this translation helpful? Give feedback.
13 replies
-
|
接上面，canal-json格式太简单，所以决定采用StreamAPI 来自定义输出的格式。现在却遇到了新的问题，麻烦大神们帮我解答一下。感谢。
自定义 import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.source.SourceRecord;
public class CustomDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
private static final long serialVersionUID = 1L;
public CustomDebeziumDeserializationSchema() {
}
public void deserialize(SourceRecord record, Collector<String> out) throws Exception {
out.collect("demo");
}
public TypeInformation<String> getProducedType() {
return BasicTypeInfo.STRING_TYPE_INFO;
}
}Stream API 用 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import com.ververica.cdc.connectors.mongodb.MongoDBSource;
public class MongoDBSourceExample {
public static void main(String[] args) throws Exception {
SourceFunction<String> sourceFunction = MongoDBSource.<String>builder()
.hosts("localhost:27017")
.username("flink")
.password("flinkpw")
.database("mydb")
.collection("orders")
.deserializer(new CustomDebeziumDeserializationSchema())
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(3000);
env.addSource(sourceFunction)
.print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
env.execute();
}
}运行报错： taskmanager 日志 命令行日志： |
Beta Was this translation helpful? Give feedback.
-
|
问题解决了。因为我写Java时间过短，长期都是写Golang，对Java的打包机制不是很熟悉。而且报错信息非常不清晰，导致我花了很久都没解决，曾一度怀疑是bug，最后在同事的帮助下解决。 记录一下过程：
1）用官方自带的
pom.xml中maven-shade-plugin configuration配置新增如下内容 <includes>
<include>org.apache.kafka:*</include>
</includes>
<relocations>
<relocation>
<pattern>org.apache.kafka</pattern>
<shadedPattern>
com.ververica.cdc.connectors.shaded.org.apache.kafka
</shadedPattern>
</relocation>
</relocations> |
Beta Was this translation helpful? Give feedback.
-
看了RowDataDebeziumDeserializeSchema.deserialize方法，代码依然不知道如何下手改，这里最终会收集一个RowData： |
Beta Was this translation helpful? Give feedback.
-
|
@leonardBang 有空解答一下么 |
Beta Was this translation helpful? Give feedback.
-
|
这个类JsonDebeziumDeserializationSchema只能在StreamApi里面用么，sql里面无法使用是吧 |
Beta Was this translation helpful? Give feedback.
Answer selected by
nienie
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2.1支持元数据列（即表名库名等信息）；但是序列化出来不是这个格式；