在线看黄aⅴ网站免费观看,亚洲中文久久精品无码91,无遮挡男女激烈动态图

您可以使用kafka-connect-tablestore包將Kafka中數(shù)據(jù)寫入Tablestore的時序表中。本文主要介紹了如何配置Kafka寫入時序數(shù)據(jù)。

前提條件

已安裝Kafka，并且已啟動ZooKeeper和Kafka。更多信息，請參見Kafka官方文檔。
已開通表格存儲服務(wù)，創(chuàng)建實例以及創(chuàng)建時序表。具體操作，請參見使用流程。
說明
您也可以通過Tablestore Sink Connector自動創(chuàng)建目標(biāo)時序表，此時需要配置auto.create為true。
已獲取AccessKey。具體操作，請參見獲取AccessKey。

背景信息

表格存儲支持對時序數(shù)據(jù)進行存儲以及分析。更多信息，請參見時序模型概述。

步驟一：部署Tablestore Sink Connector

通過以下任意一種方式獲取Tablestore Sink Connector。
- 通過GitHub下載源碼并編譯。源碼的GitHub路徑為Tablestore Sink Connector源碼。
  1. 通過Git工具執(zhí)行以下命令下載Tablestore Sink Connector源碼。
```
git clone https://github.com/aliyun/kafka-connect-tablestore.git
```
  2. 進入到下載的源碼目錄后，執(zhí)行以下命令進行Maven打包。
```
mvn clean package -DskipTests
```
    編譯完成后，生成的壓縮包（例如kafka-connect-tablestore-1.0.jar）會存放在target目錄。
- 直接下載編譯完成的kafka-connect-tablestore壓縮包。
將壓縮包復(fù)制到各個節(jié)點的$KAFKA_HOME/libs目錄下。

步驟二：啟動Tablestore Sink Connector

Tablestore Sink Connector具有standalone模式和distributed模式兩種工作模式。請根據(jù)實際選擇。

由于寫入時序數(shù)據(jù)時，Kafka側(cè)的消息記錄必須為JSON格式，因此啟動Tablestore Sink Connector時需要使用Jsonconverter，且不需要提取schema以及不需要輸入key，請在connect-standalone.properties和connect-distributed.properties中按照如下示例配置對應(yīng)配置項。

說明

如果輸入了key，請按照key的格式配置key.converter和key.converter.schemas.enable。

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

此處以配置standalone模式為例介紹，distributed模式的配置步驟與同步數(shù)據(jù)到數(shù)據(jù)表時的distributed模式配置步驟類似，只需按照上述示例在worker配置文件connect-distributed.properties中修改對應(yīng)配置項以及在connector文件connect-tablestore-sink-quickstart.json中修改時序相關(guān)配置即可。具體操作，請參見步驟二：啟動Tablestore Sink Connector中distributed模式的配置步驟。

standalone模式的配置步驟如下：

根據(jù)實際修改worker配置文件connect-standalone.properties和connector配置文件connect-tablestore-sink-quickstart.properties。

worker配置文件connect-standalone.properties的配置示例

worker配置中包括Kafka連接參數(shù)、序列化格式、提交偏移量的頻率等配置項。此處以Kafka官方示例為例介紹。更多信息，請參見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=false

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=

connector配置文件connect-tablestore-sink-quickstart.properties的配置示例

connector配置中包括連接器類、表格存儲連接參數(shù)、數(shù)據(jù)映射等配置項。更多信息，請參見配置說明。

# 設(shè)置連接器名稱。
name=tablestore-sink
# 指定連接器類。
connector.class=TableStoreSinkConnector
# 設(shè)置最大任務(wù)數(shù)。 
tasks.max=1
# 指定導(dǎo)出數(shù)據(jù)的Kafka的Topic列表。
topics=test

# 以下為Tablestore連接參數(shù)的配置。
# Tablestore實例的Endpoint。
tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
# 指定認(rèn)證模式。
tablestore.auth.mode=aksk
# 填寫AccessKey ID和AccessKey Secret。如果使用aksk認(rèn)證，則需要填入這兩項。
tablestore.access.key.id=xxx
tablestore.access.key.secret=xxx
# 指定Tablestore實例名稱。
tablestore.instance.name=xxx
## STS認(rèn)證相關(guān)配置，如果使用STS認(rèn)證，則下列各項必填。此外aksk還需要在環(huán)境變量中配置配入ACCESS_ID和ACCESS_KEY。
#sts.endpoint=
#region=
#account.id=
#role.name=

# 定義目標(biāo)表名稱的格式字符串，字符串中可包含<topic>作為原始Topic的占位符。
# topics.assign.tables配置的優(yōu)先級更高，如果配置了topics.assign.tables，則忽略table.name.format的配置。
# 例如當(dāng)設(shè)置table.name.format為kafka_<topic>時，如果kafka中主題名稱為test，則將映射到Tablestore的表名為kafka_test。
table.name.format=<topic>
# 指定Topic與目標(biāo)表的映射關(guān)系，以"<topic>:<tablename>"格式映射Topic和表名，Topic和表名之間的分隔符為半角冒號（:），不同映射之間分隔符為半角逗號（,）。
# 如果缺省，則采取table.name.format的配置。
# topics.assign.tables=test:test_kafka


# 是否自動創(chuàng)建目標(biāo)表，默認(rèn)值為false。
auto.create=true


# 以下為臟數(shù)據(jù)處理相關(guān)配置。
# 在解析Kafka Record或者寫入時序表時可能發(fā)生錯誤，您可以可通過以下配置進行處理。
# 指定容錯能力，可選值包括none和all，默認(rèn)值為none。
# none表示任何錯誤都將導(dǎo)致Sink Task立即失敗。
# all表示跳過產(chǎn)生錯誤的Record，并記錄該Record。
runtime.error.tolerance=none
# 指定臟數(shù)據(jù)記錄模式，可選值包括ignore、kafka和tablestore，默認(rèn)值為ignore。
# ignore表示忽略所有錯誤。
# kafka表示將產(chǎn)生錯誤的Record和錯誤信息存儲在Kafka的另一個Topic中。
# tablestore表示將產(chǎn)生錯誤的Record和錯誤信息存儲在Tablestore另一張數(shù)據(jù)表中。
runtime.error.mode=ignore

# 當(dāng)臟數(shù)據(jù)記錄模式為kafka時，需要配置Kafka集群地址和Topic。
# runtime.error.bootstrap.servers=localhost:9092
# runtime.error.topic.name=errors

# 當(dāng)臟數(shù)據(jù)記錄模式為tablestore時，需要配置Tablestore中數(shù)據(jù)表名稱。
# runtime.error.table.name=errors

##以下為時序表新增配置。

# connector工作模式，默認(rèn)為normal。
tablestore.mode=timeseries
# 時序表主鍵字段映射。
tablestore.timeseries.test.measurement=m
tablestore.timeseries.test.dataSource=d
tablestore.timeseries.test.tags=region,level
# 時序表時間字段映射。
tablestore.timeseries.test.time=timestamp
tablestore.timeseries.test.time.unit=MILLISECONDS
# 是否將時序數(shù)據(jù)字段（field）的列名轉(zhuǎn)為小寫，默認(rèn)為true。由于當(dāng)前時序模型中時序表的列名不支持大寫字母，如果配置為false，且列名中有大寫字母，寫入會報錯。
tablestore.timeseries.toLowerCase=true
# 是否將所有非主鍵以及時間的字段以field的形式存儲在時序表，默認(rèn)為true，如果為false，則只存儲tablestore.timeseries.test.field.name中配置的字段
tablestore.timeseries.mapAll=true
# 配置field字段名稱，多個字段名稱之間用半角冒號（,）分隔。
tablestore.timeseries.test.field.name=cpu
# 配置field字段類型。取值范圍為double、integer、string、binary和boolean。
# 當(dāng)field中包含多個字段時，字段類型必須和字段名稱一一對應(yīng)。多個字段類型之間用半角冒號（,）分隔。
tablestore.timeseries.test.field.type=double

進入到$KAFKA_HOME目錄后，執(zhí)行以下命令啟動standalone模式。

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

步驟三：生產(chǎn)新的記錄

進入到$KAFKA_HOME目錄后，執(zhí)行以下命令啟動一個控制臺生產(chǎn)者。

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

配置項說明請參見下表。

配置項	示例值	描述
--broker-list	localhost:9092	Kafka集群broker地址和端口。
--topic	test	主題名稱。啟動Tablestore Sink Connector時默認(rèn)會自動創(chuàng)建Topic，您也可以選擇手動創(chuàng)建。

向主題test中寫入一些新的消息。
重要
如果要導(dǎo)入數(shù)據(jù)到時序表，則向主題中寫入數(shù)據(jù)時必須輸入JSON格式的數(shù)據(jù)。
```
{"m":"cpu","d":"127.0.0.1","region":"shanghai","level":1,"timestamp":1638868699090,"io":5.5,"cpu":"3.5"}
```
登錄表格存儲控制臺查看數(shù)據(jù)。

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

同步數(shù)據(jù)到時序表