如何保证debezium生成的topic事件顺序，存储在kafka中并发送给spark？
要保证Debezium生成的topic事件顺序存储在Kafka中并发送给Spark，可以采取以下步骤：
- 使用Debezium进行数据变更事件的捕获和CDC（Change Data Capture）处理。Debezium是一个开源的分布式平台，用于捕获数据库的变更事件，并将其转换为可靠的流式数据流。它支持多种数据库，如MySQL、PostgreSQL、MongoDB等。
- 配置Debezium连接到目标数据库，并设置相应的CDC配置，以便捕获数据库中的变更事件。可以指定要监视的表、列等。
- 配置Debezium连接到Kafka，将捕获的变更事件作为消息发送到Kafka的topic中。可以使用Debezium提供的Kafka Connect插件来实现。
- 在Kafka中创建一个或多个topic，用于存储Debezium生成的事件。可以使用Kafka命令行工具或Kafka管理工具进行创建。
- 配置Spark连接到Kafka，订阅Debezium生成的topic，以接收事件数据。可以使用Spark Streaming或Structured Streaming来处理流式数据。
- 在Spark中编写相应的逻辑来处理接收到的事件数据。可以使用Spark的API和功能来进行数据转换、聚合、分析等操作。
通过以上步骤，可以实现Debezium生成的topic事件顺序存储在Kafka中，并通过Spark进行实时处理和分析。这种架构可以用于实时数据管道、数据集成、数据仓库等场景。
腾讯云提供了一系列与云计算相关的产品和服务，包括云数据库、消息队列、流计算等，可以用于构建类似的解决方案。具体推荐的产品和产品介绍链接地址可以根据实际需求和使用情况进行选择。
相关·内容
我用debezium得到了混乱的代码：
"doulist_name": "2013 豆瓣电影��碑榜】"
mysql数据库中有中文单词，我用debezium把数据发送给kafka。我发现在消费消息时，中文字变成了乱码，我该如何解决这个问题呢？有什么我可以使用的配置吗？
当我使用flume和kafka生成器生成中文单词时，它工作得很好。
配置的一部分：
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kaf
浏览 64提问于2018-10-16得票数 0
在我们的应用程序中，我们需要将特定postgresql表中的事件发布到Kafka中，所以我们决定使用Debezium，但遇到了以下问题:在初始快照期间，消息在Kafka中以意想不到(从我们的角度来看)的顺序显示。事件的顺序对我们的应用程序至关重要，事实上它们必须按照表的整数主键进行排序。AFAIK初始快照只是从表中选择，没有ORDER BY。那么有没有办法或者变通方法让Debezuim postgresql连接器以特定的顺序提取事件呢？ 提前感谢！
浏览 95提问于2019-10-11得票数 0
回答已采纳
我需要将MySQL数据库复制到PostgreSQL数据库。我选择了：
Debezium连接
Avro格式
合流模式注册表
卡夫卡
数据正在被复制，但是，我正在丢失一些模式信息。例如，在Postgres中将具有mysql格式的datetime格式的列复制为bigint，没有创建外键，也没有保留列的顺序(这很好)，等等。
PostgreSQL接收器连接器：
{
"name": "jdbc-sink-dbt",
"config": {
"connector.class": "
浏览 1提问于2022-01-14得票数 0
回答已采纳
我对大数据完全陌生，从最近几周开始，我一直在尝试构建日志分析应用程序。
我读了很多文章，发现Kafka + spark streaming是最可靠的配置。
现在，我能够处理从我的简单kafka java生成器发送到spark Streaming的数据了。
谁能建议几件事，如1)我如何实时读取服务器日志并将其传递给kafka broker。2)有没有从日志推送数据到Kafka的框架? 3)还有其他建议吗？？
谢谢，Chowdary
浏览 1提问于2016-05-30得票数 0
我的Dockerfile，用于下载接收器连接器和mysql jdbc驱动程序 FROM debezium/connect
ENV KAFKA_CONNECT_JDBC_DIR=$KAFKA_CONNECT_PLUGINS_DIR/kafka-connect-jdbc
ENV MYSQL_DRIVER_VERSION 5.1.39
ARG KAFKA_JDBC_VERSION=5.5.0
RUN curl -k -SL "https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-${MYSQL_DRIVER
浏览 34提问于2020-07-07得票数 0
回答已采纳
我在一个变更数据捕获项目中工作。我有一个mysql数据库。我使用debezium捕获所有更改并将其发送给kafka。后来，我读取了Spark的所有信息，并使用jdbc将其发送到Apache Phoenix。 我正在使用debezium和一个重路由选项，它只将所有表的更改发送到一个kafka主题。有了这个配置，我确信我可以按顺序阅读spark中独特的kafka主题。 但我的问题是:如果我使用debezium而没有重新路由选项，并且我在不同的kafka主题中有每个表的更改，我如何保证我以正确的顺序读取所有主题的事件？ 我知道我可以使用Spark来排序，例如按时间戳排序，但是如果比如说，一个kaf
浏览 96提问于2019-05-03得票数 0
2回答
我使用Spark2.2.0Rc1。
我有一个卡夫卡topic，我正在查询一个运行的有水印的聚合，带有一个1 minute水印，以append输出模式发送给console。
import org.apache.spark.sql.types._
val schema = StructType(StructField("time", TimestampType) :: Nil)
val q = spark.
readStream.
format("kafka").
option("kafka.bootstrap.servers",
浏览 1提问于2017-06-07得票数 15
回答已采纳
1回答
我试图在Flink中实现一个事件时时态连接。下面是第一个联接表：
tEnv.executeSql("CREATE TABLE AggregatedTrafficData_Kafka (" +
"`timestamp` TIMESTAMP_LTZ(3)," +
"`area` STRING," +
"`networkEdge` STRING," +
"`vehiclesNumber` BIGINT," +
浏览 4提问于2021-09-30得票数 0
回答已采纳
我正在使用debezium连接器读取数据写入我的数据库，然后发送给卡夫卡消费的卡夫卡消费者。我在table.include.list中指定了三个表，但它只侦听第一个表中的更改。以下是我在kafka connect中使用的配置：
connector.class=io.debezium.connector.mysql.MySqlConnector
database.user=root
topic.creation.default.partitions=3
database.server.id=184092
tasks.max=1
database.history.kafka.bootstrap.s
浏览 6提问于2021-05-05得票数 0
1回答
第一次使用Debezium时，我在debezium中每分钟只收到大约1000条消息(与在线基准相比非常慢)。没有节流卡夫卡连接/ MySQL/ Kafka经纪人，不知道我在这里做什么。我将在这里张贴配置，以供参考。
Kafka的配置-连接工作人员：
-e CONNECT_GROUP_ID="quickstart" \
-e CONNECT_CONFIG_STORAGE_TOPIC="quickstart-config" \
-e CONNECT_OFFSET_STORAGE_TOPIC="quickstart-offsets" \
-e CONN
浏览 7提问于2021-06-14得票数 1
我试图使用MongoDB debezium连接器从Kafka捕获数据，但是当我尝试用KafkaJS读取数据时，会出错：
KafkaJSProtocolError: The group member's supported protocols are incompatible with those of existing members
我正在使用码头图像来捕获数据。
以下是我要做的步骤：
启动动物园管理员
码头运行-it -rm-名动物园管理员-p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/动物园管理员:最新消息
启动卡夫卡
码头
浏览 0提问于2019-03-27得票数 0
回答已采纳
我正在构建Kafka CDC，但根据文档，它运行许多docker-run命令。
我想把它全部放到一个docker-compose.yml中，但是我在最后一个命令时失败了，我不能转换成
以下是命令
docker run -d --name postgres \
-p 5432:5432 \
-e POSTGRES_USER=start_data_engineer \
-e POSTGRES_PASSWORD=password debezium/postgres:12
docker run -d --name zookeeper
浏览 22提问于2021-12-03得票数 0
我正在尝试为mysql数据库连接多个Debezium连接器，我的配置如下。
sudo docker run -it --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.5 &
sudo docker run -it --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:1.5 &
sudo docker run -it --name connect -p 8083:8083 -e GROUP
浏览 1提问于2021-06-07得票数 0
我正在使用Azure事件集线器(kafka surface已启用)实现更改数据捕获场景，以从postgresql捕获数据。数据库是Postgresql，并使用debizum连接器()。我还使用了azure数据库(单服务器-逻辑复制)。kafka连接在本地的对接器上运行，并且可以在蔚蓝事件中心中创建主题(所附的对接者撰写文件)。另外，一旦我向kafka连接发送rest请求，它就会显示连接器已经启动并运行。但是，当我在postgres表中插入数据时，它不能在其后创建主题，而且我也无法找出原因？
连接器的状态：
localhost:8083/connectors/postgres-connecto
浏览 9提问于2022-09-22得票数 0
当我用这个堆栈设置我的Swarm时:Kafka(多代理)，zookeeper，debezium。Kafka和zookeeper正在工作，可以创建topic、consumer和producer，但debezium显示错误: org.apache.kafka.connect.errors.ConnectException:无法连接和描述Kafka集群。检查worker的代理连接和安全属性。我没有修改任何东西，只是默认配置为docker-stack：
version: '3.6'
services:
zoo:
image: wurstmeister/zookeepe
浏览 23提问于2018-08-03得票数 0
我正在尝试为需要运行多个docker容器的CDC (变更数据捕获)解决方案创建一个docker-compose文件。我的DockerComposeLocal.yml包含以下内容： version: "3"
services:
postgres:
image: debezium/postgres:12
ports:
- "5432:5432"
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=123
zookeper:
浏览 16提问于2021-08-16得票数 0
我们在Server实例上有一个始终不变的实例。我想用Debezium在一些桌子上跟踪CDC事件。通过遵循debezium网站上的官方教程，我注册了一个具有以下属性的连接器：
{
"name": "debezium-alwayson-connector",
"config": {
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"database.hostname&
浏览 3提问于2021-11-29得票数 0
回答已采纳
1回答
我的问题是：
version: '2'
services:
zookeeper:
container_name: zookeeper
image: debezium/zookeeper:${DEBEZIUM_VERSION}
ports:
- 2181:2181
- 2888:2888
- 3888:3888
kafka:
container_name: kafka
image: debezium/kafka:${DEBEZIUM_VERSION}
ports:
- 9092:909
浏览 3提问于2021-04-26得票数 0
我想知道如何获得来自Debezium sql server连接器的所有日期时间(io.debezium.time.Timestamp)，格式为YYYY-MM-DD hh:mm:ss。现在，这是以纪元int64格式给出的，这将很难使用spark转换数据。 下面使用的debezium sql server连接器的配置： {
"name": "localDB-sqlserverconnector",
"config": {
"connector.class": "io.debezium.conn
浏览 52提问于2020-10-01得票数 1
1回答
我计划使用向Kafka生成事件，我需要强大的交付和订购保证。通过在生产者配置中使用enable.idempotence=true参数，我可以得到这些保证。
我的问题是：
我怎么能用debezium来实现这一点呢？
这些担保是默认提供的吗？
是可配置的吗？多么?
注意:我计划使用
浏览 1提问于2020-02-09得票数 4
回答已采纳
在PostgreSQL支付(模式)发件箱(表)上配置为Kafka主题(支付-事务)。
Means at Kafka Connect perform below actions
payments.outbox ==streams=> payments-transactions Topic
作为新CDC流程的一部分，我们计划将webhooks(schema)发件箱(表)配置为不同的Kafka主题(Webhooks)。
Means at Kafka Connect perform below actions
payments.outbox ==streams=> payments-tra
浏览 1提问于2022-08-10得票数 0
我有一个debezium实例，它跟踪postgres上的更改并将更改消息发送给kafka。在kafka的另一端，我使用了带有amazon redshift sink connector的kafka connect，连接器所期望的消息格式与debezium发送给kafka的消息格式不兼容。有没有办法强迫debezium发送amazon redshift sink connector可以接受的消息？
浏览 2提问于2021-12-03得票数 0
debezium kafka连接命令是：docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:0.9
Pl
浏览 3提问于2019-06-05得票数 0
比方说，我将获得多个MySQL实例，并且必须流式传输实时数据。为此，我使用下面列出的命令在Docker Container上启动了一些服务，这些命令将我连接到示例库存数据库，并且当我在MySQL命令行客户端中执行任何DML命令时，都能够监控这些数据的变化。
但是，连接到实际的MySQL数据库并在观察器窗口中传输相同的内容所需的更改或等效命令是什么呢？
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.1
docker run -it --rm --n
浏览 2提问于2020-05-27得票数 1
1回答
我遵循了以下步骤：
# Run postgres instance
docker run --name postgres -p 5000:5432 debezium/postgres
# Run zookeeper instance
docker run -it --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper
# Run kafka instance
docker run -it --name kafka -p 9092:9092 --link zookeeper:zookeeper de
浏览 6提问于2019-08-02得票数 0
debezium kafka connect命令是：docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect
浏览 21提问于2019-06-05得票数 1
回答已采纳
我成功地安装了Postgres Debezium CDC。现在，我能够捕获数据库中发生的所有更改。但问题是“之前”字段始终保持为空。因此，如果我插入一个记录(id = 1, name = Bill)，然后从Kafka获得以下数据：
'payload': {'before': None, 'after': {'id': 1, 'name': 'Bill'}, ...
但如果我像这样更新记录：
UPDATE mytable set name = 'Bob' WHERE id = 1
我从
浏览 0提问于2020-01-18得票数 5
回答已采纳
1回答
我使用了debezium.io 提供的教程，并将其调整为使用MySQL数据库而不是MySQL，但是当我在数据库中进行更改时，观察者不会显示任何事件或活动。以下是我所采取的步骤：
我运行了动物园管理员码头命令：
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.1
然后我运行了kafka停靠命令：
docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper de
浏览 4提问于2020-06-12得票数 2
回答已采纳
2回答
我试着用Debezium和卡夫卡连接，我跟踪，所有连接都很好。然而，问题是，我不能再从码头集装箱以外的卡夫卡。
我使用以下命令启动容器：
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:2.0.0.Beta1
docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:2.0.0.Beta1
docker run -it --rm --
浏览 3提问于2022-08-14得票数 0
回答已采纳
在此之前，我已经在SQL Server中基于此设置了CDC，并基于此将Debezium SQL Server连接器注册到Kafka connect。我运行Kafka消费者脚本，在SQL Server中执行一些修改数据(插入、更新和删除)，但Kafka消费者没有显示任何关于修改数据的JSON日志。
这是我的配置文件sqlserver-debezium.json：
{
"name":"sqlserver-debezium",
"config":{
"connector.class":"io.debezi
浏览 58提问于2020-05-05得票数 0
尝试将Debezium任务提交给confluent抛出以下异常。尝试手动创建主题，然后提交任务，仍然得到相同的异常
org.apache.kafka.connect.errors.ConnectException: Creation of database history topic failed, please create the topic manually
at io.debezium.relational.history.KafkaDatabaseHistory.initializeStorage(KafkaDatabaseHistory.java:349)
at io.debezi
浏览 28提问于2018-08-02得票数 0
我使用Apache Kafka和Debezium的连接器服务来查看我的表中的数据更改。另外，我使用的是亚马逊网络服务EC2，其中我有一个包含联邦表的数据库。如果我创建了一个非联邦表，并使用Zookeeper、Kafka和Kafka连接来运行我的docker观察器，我可以从该表中读取数据，但是当我看到一个联邦表时，就不能读取数据。 我试图在连接器的配置中添加一个查询，但这不起作用。 我的命令是： 启动Zookeeper $ docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/z
浏览 20提问于2019-07-02得票数 0
我想知道这些命令之间的区别。 -- With bootstrap server
kafka-topics \
--bootstrap-server b-1.bhuvi-cluster-secure.jj9mhr.c3.kafka.ap-south-1.amazonaws.com:9098,b-2.bhuvi-cluster-secure.jj9mhr.c3.kafka.ap-south-1.amazonaws.com:9098 \
--delete \
--topic debezium-my-topic \
--command-config /etc/kafka/client.prope
浏览 24提问于2021-08-26得票数 0
回答已采纳
1回答
对接MySQL数据库
、、、、
我是新来的debezium，正在尝试将我的kafka连接器连接到我在本地WAMP服务器上已经存在的MySQL数据库。
根据教程文档，我开始了动物园管理员和kafka，然后我启动了kafka连接器。在我的邮递员中，我向我的kafka连接器发送了以下JSON，但是我一直收到一个糟糕的请求响应。
启动Zookepper
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:0.7
启动卡夫卡
docker run -it --rm --name kafka -p 9
浏览 0提问于2018-06-21得票数 0
我是卡夫卡的新手，我不知道如何使用"transforms.router.type“来使它与我的Debezium设置一起工作。因此，我对java类进行了特殊的事件转换，并对其进行了预配置，以便将其部署到容器中，如下所示：
curl -X POST -H "Accept:application/json“-H”内容-Type:application/json“localhost:8083/连接器/ -d
{
"name": "task-connector",
"config": {
"connector.cla
浏览 6提问于2020-08-07得票数 1
回答已采纳
我使用的是lyft flink operator部署的Flink 1.14 我正在尝试使用Table API使tumble窗口聚合，从transactions表源读取，并将窗口的聚合结果放入新的kafka主题中 我的资料来源是来自debezium的卡夫卡话题。 EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tEnv = TableEnvironment.create(settings);
//this is the source
tEnv.exec
浏览 126提问于2021-11-14得票数 0
回答已采纳
我是新来的debezium，我找不到关于用特定密钥从数据库向kafka发送消息的信息。是否可以根据主键之一配置debzium来发送具有特殊密钥的消息？我可以使用debezium变换或一些正则表达式来实现这一点吗？
浏览 1提问于2021-12-21得票数 0
2回答
问题：
如何使用Kafka正确注册SqlServer连接器以连接到独立SqlServer实例？注意:我没有在Docker中运行Server。
错误：
com.microsoft.sqlserver.jdbc.SQLServerException:错误:由：连接到主机的/IP连接127.0.0.1，端口1433失败。错误：“连接拒绝(连接拒绝)。请验证连接属性。确保Server实例在主机上运行，并在端口接受TCP/IP连接。确保到端口的TCP连接不会被防火墙阻塞”。
Troubleshooting:
在Server配置管理器的Server ConfigurationDisabled Firew
浏览 4提问于2020-04-07得票数 1
回答已采纳
我面临的问题是通过CDC使用带有mysql的Debezium连接器从mysql表中获得双值。
以下是连接器道具：
{
"name": "test",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "10",
"database.hostname": "mysql.default",
浏览 4提问于2021-03-12得票数 0
回答已采纳
2回答
我是kubernetes的新手，我正在尝试将kafka与debezium和mysql集成。我成功地在minikube上部署了kafka和mysql，一旦我在minikube上部署了debezium yml，它就被挂起了，完全没有响应，然后我重新启动minikube，在运行完所有的pod minikube之后再次被挂起。下面是我的代码：
zookeeper服务
apiVersion: v1
kind: Service
metadata:
name: zoo1
labels:
app: zookeeper-1
spec:
ports:
- name: client
浏览 2提问于2018-12-08得票数 0
我设计了一个Nifi流，将以Avro格式序列化的JSON事件推送到Kafka topic中，然后尝试在Spark Structured streaming中消费它。
虽然Kafka part运行良好，但Spark Structured streaming无法读取Avro事件。它失败，并出现以下错误。
[Stage 0:> (0 + 1) / 1]2019-07-19 16:56:57 ERROR Utils:91 - Aborting task
org.apache.avro
浏览 19提问于2019-07-19得票数 0
我不能使用以下命令添加MySQL连接器： curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "useraccount-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector
浏览 56提问于2020-10-15得票数 2
我当前的测试配置如下所示：
version: '3.7'
services:
postgres:
image: debezium/postgres
restart: always
ports:
- "5432:5432"
zookeeper:
image: debezium/zookeeper
ports:
- "2181:2181"
- "2888:2888"
- "3888:3888"
kafka:
im
浏览 4提问于2020-01-15得票数 2
回答已采纳
我使用Debezium作为CDC工具来流式传输来自MySql的数据。在将Debezium连接器安装到Confluent OSS集群后，我正在尝试将MySQL bin_log的变化捕获到Kafka主题中。当我创建连接器时，在获取数据库的快照之后，我会遇到一系列连续的错误。
我检查了MySql bin_log是否打开，并尝试使用不同的序列化程序重新启动模式注册表和连接器。但是我得到了同样的错误。
错误日志显示：
[2019-06-21 13:56:14,885] INFO Step 8: - Completed scanning a total of 955 rows from table
浏览 1提问于2019-06-21得票数 3
2回答
有人能告诉我，在osgi包中使用嵌入式Debezium (没有kafka)和使用Debezium与kafka有什么区别？我在网上找不到多少关于Debezium的东西。
浏览 3提问于2017-04-10得票数 1
回答已采纳
我使用过Debezium作为Mysql的-> Elasticsearch。现在的问题是，当我从MySQL中删除数据时，它仍然会在Elasticsearch中重新出现，即使数据不再存在于MySQL DB中。UPDATE和INSERT工作正常，但DELETE不是。另外，我做了以下工作：
Delete在MySQL中的数据
Delete Elasticsearch索引和ES Kafka Sink
Create -- Kakfa中ES的新连接器
现在，奇怪的是，所有我删除的数据收割机在这里，以及！当我在步骤(3)之前检查ES数据时，数据不存在。但之后，这种行为会被观察到。
请帮我解决这个问题！
M
浏览 7提问于2021-03-14得票数 0
我正在创建一个CDC Postgres.source --> Kafka / Debezium --> Postgres.sink。 源主题创建得很好，但只有在表有数据的情况下才能创建。有没有办法在没有数据的情况下实例化所有的主题？因为我想用maven命令下载所有AVRO模式： mvn schema-registry:download 我的配置： {
"name": "postgres-source",
"config": {
"connector.class":"io.debezium.co
浏览 20提问于2021-07-30得票数 0
2回答
我正在尝试使用Debezium连接到我本地机器上的mysql数据库。
尝试使用以下命令调用kafka: sudo kafka/bin/connect-standalone.shsh kafka/config/connect-standalone.properties kafka/config/connector.properties
下面是connector.properties中的配置：
{
"name": "inventory-connector",
"config": {
"connector.class":
浏览 0提问于2018-12-07得票数 3
我正在读取镶木地板文件，并将其转换为JSON格式，然后发送给kafka。问题是，它读取了整个拼图，所以一次性发送给kafka，但我想逐行或分批发送json数据： object WriteParquet2Kafka {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession
.builder
.master("yarn")
.appName("Write Parquet to Kafka")
.getO
浏览 18提问于2019-01-22得票数 1
回答已采纳
1回答
我正在尝试为MySQL数据库中的多个表配置Debezium连接器(在MySQL 8.0上使用debezium 1.4 )。我的公司在用kafka创建主题时有一个命名模式可遵循，而且这种模式不允许使用下划线(_)，所以我不得不用连字符(-)替换它们。
因此，我的主题名称是：
主题1
fjf.db.top-domain.domain.sub-domain.transaction-search.order-status
WHERE
- transaction-search = schema "transaction_search"
- order-status = table
浏览 5提问于2021-02-17得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
