Vertica 7.2 卡住从 Kafka 导入消息答案

【问题标题】：Vertica 7.2 stuck importing messages from KafkaVertica 7.2 卡住从 Kafka 导入消息
【发布时间】：2015-11-09 09:28:11
【问题描述】：

我正在尝试在 Vertica v7.2.0-0 新功能中处理 Kafka。我正在运行 Kafka 代理，主题“事件”与 json 数据。我在 Vertica 中配置了 Kafka 调度程序，设置了代理、目标表、Kafka JSON 解析器等。一切都很好，除非 Vertica 收到损坏的消息，她回滚 COPY 事务并且不移动偏移量而不是获取新消息，只是跳过错误或写入em 到被拒绝的表。假设我在表中有 NOT NULL 列，当 Vertica 获取 json 而没有需要的列时，下一个错误连续发生（来自 vertica.log）：

COPY "public"."events" SOURCE KafkaSource(stream='events|0|410015', brokers='1.2.3.4:9092', duration=interval '9884 milliseconds', stop_on_eof=true, executionparallelism=1 ) PARSER KafkaJSONParser( ) REJECTED DATA AS TABLE public.kafka_rej2 TRICKLE NO COMMIT 
Cannot set a NOT NULL column (device_type) to a NULL value in COPY statement

以及它在 kafka_config.kafka_events 中的外观：

发生错误，从微批处理回滚更改。
| java.sql.SQLDataException：[Vertica]VJDBC 错误：无法设置在 COPY 语句中将 NOT NULL 列 (device_type) 转换为 NULL 值 com.vertica.util.ServerErrorData.buildException（未知来源）在 com.vertica.dataengine.VResultSet.fetchChunk（未知来源）在 com.vertica.dataengine.VResultSet.initialize（未知来源）在 com.vertica.dataengine.VQueryExecutor.readExecuteResponse（未知来源）在 com.vertica.dataengine.VQueryExecutor.handleExecuteResponse（未知来源）在 com.vertica.dataengine.VQueryExecutor.execute（未知来源）在 com.vertica.jdbc.common.SStatement.executeNoParams（未知来源）在 com.vertica.jdbc.common.SStatement.executeUpdate（未知来源）在 com.vertica.solutions.kafka.scheduler.MicroBatch.execute(MicroBatch.java:158) 在 com.vertica.solutions.kafka.scheduler.LaneWorker.run(LaneWorker.java:67) 在 java.lang.Thread.run(Unknown Source) 引起： com.vertica.support.exceptions.DataException：[Vertica]VJDBC 错误：无法将 NOT NULL 列 (device_type) 设置为 NULL 值 COPY 声明... 11 更多

忘记了我有错误，但是如何告诉 Vertica 跳过复制错误并转到 Kafka 主题中的下一个偏移量？我已经尝试了所有 KafkaJSONParser 选项，但没有效果。

【问题讨论】：

我觉得这个问题会更适合官方HPE forum。

标签： apache-kafka vertica

【解决方案1】：

我相信这是 COPY 和约束与 Vertica 一起使用的一种人工产物。通常当记录是“坏的”时，COPY 会拒绝该记录并继续前进。但是，NOT NULL 约束发生在加载堆栈的更下方，并导致整个语句回滚。因此，偏移计数器永远不会增加：这会导致数据丢失。

目前的解决方法是定位一个没有 NOT NULL 约束的表。

【讨论】：

【解决方案2】：

您必须创建弹性表而不是典型表。尝试以下命令（在 COPY 命令之前）：

CREATE FLEX TABLE events();

【讨论】：