当前位置:网站首页>Kafka connect synchronizes Kafka data to MySQL
Kafka connect synchronizes Kafka data to MySQL
2022-07-08 01:31:00 【W_ Meng_ H】
One 、 Background information
Kafka Connect It is mainly used to input and output data streams to message queues Kafka edition .Kafka Connect Mainly through various Source Connector The implementation of the , Input data from a third-party system into Kafka Broker, Through a variety of Sink Connector Realization , Take data from Kafka Broker Import into the third-party system .
Official documents :How to use Kafka Connect - Getting Started | Confluent Documentation
Two 、 development environment
middleware | edition |
zookeeper | 3.7.0 |
kafka | 2.10-0.10.2.1 |
3、 ... and 、 To configure Kafka Connect
1、 Enter kafka Of config Under the folder , modify connect-standalone.properties
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# These are defaults. This file just demonstrates how to override some settings.
bootstrap.servers=kafka-0:9092
# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
# need to configure these based on the format they want their data in when loaded from or stored into Kafka
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
# it to
key.converter.schemas.enable=false
value.converter.schemas.enable=false
# The internal converter used for offsets and config data is configurable and must be specified, but most users will
# always want to use the built-in default. Offset and config data is never visible outside of Kafka Connect in this format.
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000
# The core configuration
#plugin.path=/home/kafka/plugins
If Mysql Data synchronization to Kafka, The following information needs to be modified :
# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
# need to configure these based on the format they want their data in when loaded from or stored into Kafka
key.converter=io.confluent.connect.json.JsonSchemaConverter
value.converter=io.confluent.connect.json.JsonSchemaConverter
# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
# it to
key.converter.schemas.enable=true
value.converter.schemas.enable=true
Be careful :Kafka Connect The previous version of does not support configuration plugin.path, You need to CLASSPATH Specify the plug-in location in
vi /etc/profile
export CLASSPATH=/home/kafka/*
source /etc/profile
2、 modify connect-mysql-source.properties(mysql-kafka)
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Confluent Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# http://www.confluent.io/confluent-community-license
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#
# A simple example that copies all tables from a SQLite database. The first few settings are
# required for all connectors: a name, the connector class to run, and the maximum number of
# tasks to create:
name=test-source-mysql
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
# The remaining configs are specific to the JDBC source connector. In this example, we connect to a
# SQLite database stored in the file test.db, use and auto-incrementing column called 'id' to
# detect new rows as they are added, and output to topics prefixed with 'test-sqlite-jdbc-', e.g.
# a table called 'users' will be written to the topic 'test-sqlite-jdbc-users'.
connection.url=jdbc:mysql://localhost:3306/demo?user=root&password=root
table.whitelist=test
mode=incrementing
incrementing.column.name=id
topic.prefix=mysql-
# Define when identifiers should be quoted in DDL and DML statements.
# The default is 'always' to maintain backward compatibility with prior versions.
# Set this to 'never' to avoid quoting fully-qualified or simple table and column names.
#quote.sql.identifiers=always
Related drive jar package :
kafka- Related driver package -Java Document resources -CSDN download
3、 modify connect-mysql-sink.properties(kafka-mysql)
#
# Copyright 2018 Confluent Inc.
#
# Licensed under the Confluent Community License (the "License"); you may not use
# this file except in compliance with the License. You may obtain a copy of the
# License at
#
# http://www.confluent.io/confluent-community-license
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
#
# A simple example that copies from a topic to a SQLite database.
# The first few settings are required for all connectors:
# a name, the connector class to run, and the maximum number of tasks to create:
name=test-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
# The topics to consume from - required for sink connectors like this one
topics=mysql-test_to_kafka
# Configuration specific to the JDBC sink connector.
# We want to connect to a SQLite database stored in the file test.db and auto-create tables.
connection.url=jdbc:mysql://localhost:3306/demo?user=root&password=root
auto.create=false
pk.mode=record_value
pk.fields=id
table.name.format=test_kafka_to
#delete.enabled=true
# Write mode
insert.mode=upsert
# Define when identifiers should be quoted in DDL and DML statements.
# The default is 'always' to maintain backward compatibility with prior versions.
# Set this to 'never' to avoid quoting fully-qualified or simple table and column names.
#quote.sql.identifiers=always
Four 、 Start command
bin/connect-standalone.sh config/connect-standalone.properties config/connect-mysql-source.properties config/connect-mysql-sink.properties
边栏推荐
- 4. Apprentissage stratégique
- Redis 主从复制
- Two methods for full screen adaptation of background pictures, background size: cover; Or (background size: 100% 100%;)
- break net
- Redis cluster
- 写一个纯手写的qt的hello world
- High quality USB sound card / audio chip sss1700 | sss1700 design 96 kHz 24 bit sampling rate USB headset microphone scheme | sss1700 Chinese design scheme explanation
- Led serial communication
- AttributeError: ‘str‘ object has no attribute ‘strftime‘
- Grey correlation analysis link (portal) matlab
猜你喜欢
COMSOL----微阻梁模型的搭建---最终的温度分布和变形情况---材料的添加
3、多智能体强化学习
Application of state mode in JSF source code
Ag9310 same function alternative | cs5261 replaces ag9310type-c to HDMI single switch screen alternative | low BOM replaces ag9310 design
2021 welder (primary) examination skills and welder (primary) operation examination question bank
Ag9310 design USB type C to hdmi+u2+5v slow charging scheme design | ag9310 expansion dock scheme circuit | type-C dongle design data
2022 high voltage electrician examination skills and high voltage electrician reexamination examination
Ag9310 for type-C docking station scheme circuit design method | ag9310 for type-C audio and video converter scheme circuit design reference
Blue Bridge Cup embedded (F103) -1 STM32 clock operation and led operation method
2022 safety officer-c certificate examination summary and safety officer-c certificate reexamination examination
随机推荐
5. Discrete control and continuous control
Scalar / vector / matrix derivation method
Probability distribution
Generic configuration legend
Scheme selection and scheme design of multifunctional docking station for type C to VGA HDMI audio and video launched by ange in Taiwan | scheme selection and scheme explanation of usb-c to VGA HDMI c
Redis cluster
2021 tea master (primary) examination materials and tea master (primary) simulation test questions
Introduction to the types and repair methods of chip Eco
NPM Internal Split module
5. Contrôle discret et contrôle continu
Common operations of numpy on two-dimensional array
Talk about smart Park
2022 R1 fast opening pressure vessel operation test question bank and R1 fast opening pressure vessel operation free test questions
Write a pure handwritten QT Hello World
Understanding of expectation, variance, covariance and correlation coefficient
4. Strategic Learning
2022 free test questions of fusion welding and thermal cutting and summary of fusion welding and thermal cutting examination
Measure the voltage with analog input (taking Arduino as an example, the range is about 1KV)
About snake equation (1)
Multi purpose signal modulation generation system based on environmental optical signal detection and user-defined signal rules