JDBC Teradata Sink：如何插入空值答案

【问题标题】：JDBC Teradata Sink: How to insert null valuesJDBC Teradata Sink：如何插入空值
【发布时间】：2019-10-20 22:14:23
【问题描述】：

我正在尝试将数据从 Kafka 插入 Teradata。有效负载有一些空值，并且 JDBC 接收器抛出以下错误。

[Teradata JDBC Driver] [TeraJDBC 16.20.00.10] [Error 1063] [SQLState HY000] null is not supported as a data value with this variant of the setObject method; use the setNull method or the setObject method with a targetSqlType parameter

我的连接器配置：

name=teradata-sink-K_C_OSUSR_DGL_DFORM_I1-V2
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
connection.url=connectionString
topics=POPS-P-OSUSR_DGL_DFORM_I1-J-V2-CAL-OUT
topic.prefix=
table.name.format=K_C_OSUSR_DGL_DFORM_I1_V2
batch.size=50000
errors.tolerance=all
errors.deadletterqueue.topic.name=POPS-P-OSUSR_DGL_DFORM_I1-V2-CAL-DEAD
errors.deadletterqueue.topic.replication.factor=1

有没有办法做到这一点？我不知道我是否必须将一些代码更改为接收器或只是更改连接器配置。

【问题讨论】：

你使用的是哪个 Kafka 版本？
另外，能分享一下连接器的配置吗？
@GiorgosMyrianthous 完成，我认为版本是 3.2.2
您的目标架构是什么样的？您的源消息是什么样的？

标签： jdbc apache-kafka teradata apache-kafka-connect

【解决方案1】：

您从某些看起来像这样的行中得到错误：

ps.setObject(1, val);

如果您尝试插入的val 具有null 值，则此选项将引发异常。

错误告诉您必须指定传入的空值的数据类型。你可以这样做：

ps.setObject(1, val, Types.VARCHAR);

这样您将NULL 转换为VARCHAR，这是受支持的targetSqlTypes 之一。

出于相同目的的另一个选择：

ps.setNull(1, Types.VARCHAR) ;

【讨论】：

所以我需要更改 JDBC Kafka Sink 中的那一行？
我真的不知道那个从未使用过的水槽的细节，但我想你应该知道。这里有更多信息，以及实现此目的的另一种方法：community.teradata.com/t5/Third-Party-Software/…
接收器连接器使用 JDBC。这里的答案取决于源消息是否与目标模式匹配。如果是这样，那么 Teradata 的连接器应该有一个新的方言 (github.com/confluentinc/kafka-connect-jdbc/tree/master/src/main/…)，专门用于处理任何数据转换问题。如果架构不匹配，那么对齐它们应该可以工作，并且不需要任何实际的代码更改。

【解决方案2】：

我们遇到的问题是我们使用标准的 Kafka Connect 来创建接收器（我们没有编写任何自定义连接器）。

我们已经为 worker 和 connector 配置了 .properties 文件，以在主题和 teradata 表之间创建链接，并使用它运行它

.../confluent/bin/connect-standalone <worker.cfg> <connector.cfg>

当我们使用“null”值创建消息并将其发送到主题时，Sink 连接器无法将记录插入到 TD 表中。

【讨论】：