Tablestore:Sink data Kafka ke tabel data

• Contoh file konfigurasi worker connect-standalone.properties

  Konfigurasi worker mencakup parameter seperti pengaturan koneksi Kafka, format serialisasi, dan frekuensi commit offset. Contoh ini berdasarkan dokumentasi resmi Kafka. Untuk informasi lebih lanjut, lihat Kafka Connect.

  # 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=localhost: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=true
  value.converter.schemas.enable=true
  
  offset.storage.file.filename=/tmp/connect.offsets
  # Flush much faster than normal, which is useful for testing/debugging
  offset.flush.interval.ms=10000
  
  # Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
  # (connectors, converters, transformations). The list should consist of top level directories that include 
  # any combination of: 
  # a) directories immediately containing jars with plugins and their dependencies
  # b) uber-jars with plugins and their dependencies
  # c) directories immediately containing the package directory structure of classes of plugins and their dependencies
  # Note: symlinks will be followed to discover dependencies or plugins.
  # Examples: 
  # plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
  #plugin.path=

• Contoh file konfigurasi konektor: connect-tablestore-sink-quickstart.properties

  Konfigurasi konektor mencakup parameter seperti kelas konektor, koneksi Tablestore, dan pemetaan data. Untuk informasi lebih lanjut, lihat Konfigurasi.

  # Set the connector name.
  name=tablestore-sink
  # Specify the connector class.
  connector.class=TableStoreSinkConnector
  # Set the maximum number of tasks.
  tasks.max=1
  # Specify the list of Kafka topics from which to export data.
  topics=test
  
  # The following are the Tablestore connection parameter settings.
  # The endpoint of the Tablestore instance.
  tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
  # Your AccessKey ID and AccessKey secret.
  tablestore.access.key.id =xxx
  tablestore.access.key.secret=xxx
  # The Tablestore instance name.
  tablestore.instance.name=xxx
  
  # A format string for the destination Tablestore table name. <topic> is a placeholder for the original topic name. The default value is <topic>.
  # Examples:
  # table.name.format=kafka_<topic>, message records from the 'test' topic will be written to the 'kafka_test' data table.
  # table.name.format=
  
  # The primary key mode. The default value is kafka.
  # The primary key of the Tablestore data table will be composed of <topic>_<partition> (the Kafka topic and partition, separated by "_") and <offset> (the offset of the message record in the partition).
  # primarykey.mode=
  
  # Automatically create the destination table. The default value is false.
  auto.create=true

bin/connect-standalone.sh config/connect-standalone.properties config/connect-tablestore-sink-quickstart.properties

Konfigurasi worker mencakup parameter seperti pengaturan koneksi Kafka, format serialisasi, dan frekuensi commit offset. Konfigurasi ini juga menentukan topik yang menyimpan informasi untuk setiap konektor. Anda harus membuat topik-topik tersebut terlebih dahulu. Contoh ini berdasarkan dokumentasi resmi Kafka. Untuk informasi lebih lanjut, lihat Kafka Connect.

• offset.storage.topic: Topik compact yang digunakan untuk menyimpan offset setiap konektor.

• config.storage.topic: Topik compact yang digunakan untuk menyimpan konfigurasi konektor dan task. Jumlah partisi untuk topik ini harus diatur menjadi 1.

• status.storage.topic: Topik compact yang digunakan untuk menyimpan informasi status Kafka Connect.

##
# 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.
##

# This file contains some of the configurations for the Kafka Connect distributed worker. This file is intended
# to be used with the examples, and some settings may differ from those used in a production system, especially
# the `bootstrap.servers` and those specifying replication factors.

# A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
bootstrap.servers=localhost:9092

# unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs
group.id=connect-cluster

# 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=true
value.converter.schemas.enable=true

# Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
#offset.storage.partitions=25

# Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated,
# and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
config.storage.topic=connect-configs
config.storage.replication.factor=1

# Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted.
# Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create
# the topic before starting Kafka Connect if a specific topic configuration is needed.
# Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value.
# Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able
# to run this example on a single-broker cluster and so here we instead set the replication factor to 1.
status.storage.topic=connect-status
status.storage.replication.factor=1
#status.storage.partitions=5

# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000

# These are provided to inform the user about the presence of the REST host and port configs 
# Hostname & Port for the REST API to listen on. If this is set, it will bind to the interface used to listen to requests.
#rest.host.name=
#rest.port=8083

# The Hostname & Port that will be given out to other workers to connect to i.e. URLs that are routable from other servers.
#rest.advertised.host.name=
#rest.advertised.port=

# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
# (connectors, converters, transformations). The list should consist of top level directories that include 
# any combination of: 
# a) directories immediately containing jars with plugins and their dependencies
# b) uber-jars with plugins and their dependencies
# c) directories immediately containing the package directory structure of classes of plugins and their dependencies
# Examples: 
# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
#plugin.path=

bin/connect-distributed.sh config/connect-distributed.properties