更改与 HIVE 集成的 HDFS2 SINK CONNECTOR 中的主题名称答案

【问题标题】：Change the topic name in a HDFS2 SINK CONNECTOR integrated with HIVE更改与 HIVE 集成的 HDFS2 SINK CONNECTOR 中的主题名称
【发布时间】：2021-01-19 16:35:30
【问题描述】：

早安

当我使用与 HIVE 集成的 HDFS2 连接器接收器时，数据库表会获取主题的名称。有没有办法选择表的名称？

这是我的连接器的配置：

   "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
   "hive.integration": "true",
   "hive.database": "databaseEze",
   "hive.metastore.uris": "thrift://server1.dc.es.arioto:9083",
   "transforms.InsertField.timestamp.field": "carga",
   "flush.size": "100000",
   "tasks.max": "2",
   "timezone": "Europe/Paris",
   "transforms": "RenameField,InsertField,carga_format",
   "rotate.interval.ms": "900000",
   "locale": "en-GB",
   "logs.dir": "/logs",    
   "format.class": "io.confluent.connect.hdfs.avro.AvroFormat",
   "transforms.InsertField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
   "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
   "value.converter": "io.confluent.connect.avro.AvroConverter",
   "key.converter": "org.apache.kafka.connect.storage.StringConverter",
   "transforms.RenameField.renames": "var1:Test1,var2:Test2,var3:test3",
   "transforms.carga_format.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
   "transforms.carga_format.target.type": "string",
   "transforms.carga_format.format": "yyyyMMdd",
   "hadoop.conf.dir": "/etc/hadoop/",
   "schema.compatibility": "BACKWARD",
   "topics": "Skiel-Tracking-Replicator",
   "hdfs.url": "hdfs://database/user/datavaseEze/",
   "transforms.InsertField.topic.field": "ds_topic",    
   "partition.field.name": "carga",
   "transforms.InsertField.partition.field": "test_partition",
   "value.converter.schema.registry.url": "http://schema-registry-eze-dev.ocjc.serv.dc.es.arioto",
   "partitioner.class": "io.confluent.connect.storage.partitioner.FieldPartitioner",
   "name": "KAFKA-HDFS-HIVE-TEST",
   "transforms.fx_carga_format.field": "carga",
   "transforms.InsertField.offset.field": "test_offset"
}

With that config, the table will name **Skiel-Tracking-Replicator** and I want that the table name will be d9nvtest.

【问题讨论】：

标签： apache-kafka hive hdfs apache-kafka-connect confluent-platform

【解决方案1】：

您可以使用RegexRouter Single Message Transform 来修改主题名称。

{
"transforms"                        : "renameTopic",
"transforms.renameTopic.type"       : "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.renameTopic.regex"      : "Skiel-Tracking-Replicator",
"transforms.renameTopic.replacement": "d9nvtest"
}

见https://rmoff.net/2020/12/11/twelve-days-of-smt-day-4-regexrouter/

【讨论】：

【解决方案2】：

在使用带有 kafka-connect-hdfs 的 RegexRouter 时，会出现此问题 - https://github.com/confluentinc/kafka-connect-hdfs/issues/236

这里的最后一条评论指出这两者在概念上是不兼容的。

【讨论】：