Load data to Vertica regular table through Kafka

I notice that you are using kafkaParser and by default it is considered as single message. You need specify the format of the data and use a parser such as JSON , Avro etc. Do you see any data in the table test_rejected?

https://my.vertica.com/docs/9.1.x/HTML/index.htm#Authoring/KafkaIntegrationGuide/KafkaFunctions/KafkaParser.htm

maverdb2=> select * from test_rejected;
ERROR 4566: Relation "test_rejected" does not exist

I seems like this table doesn't exist.

My data is in text format with delimiter "|". Can I use KafkaJSONParser to parse this regular text format?

So it means that txn did not succeed. I sent you a personal message. please send me vertica.log to check what happened to that transaction

I saw Vertica is doing the following in the log

COPY "public"."test" SOURCE KafkaSource(stream='oggtopic|0|115', brokers='testserver01.production.local:9092', duration=interval '9871 milliseconds', eof_timeout=interval '1000 milliseconds', stop_on_eof=true, message_max_bytes=1048576 ) PARSER KafkaParser() TRICKLE STREAM NAME 'vkstream_verget_microbatch_500001_orafeed' NO COMMIT

I run this command in vsql, and nothing was loaded. I played with following.

so it looks like if I remove PARSER KafkaParser(), it works. How to unattach the PARSER KafkaParser() from the load_spec? or what is the standard native parser for pure text?

maverdb2=> COPY "public"."test" SOURCE KafkaSource(stream='oggtopic|0|115', brokers='testserver01.production.local:9092', duration=interval '9871 milliseconds', eof_timeout=interval '1000 milliseconds', stop_on_eof=true, message_max_bytes=1048576 ) STREAM NAME 'vkstream_verget_microbatch_500001_orafeed' NO COMMIT;

Rows Loaded
2
(1 row)

maverdb2=> COPY "public"."test" SOURCE KafkaSource(stream='oggtopic|0|115', brokers='testserver01.production.local:9092', duration=interval '9871 milliseconds', eof_timeout=interval '1000 milliseconds', stop_on_eof=true, message_max_bytes=1048576 ) PARSER KafkaParser() STREAM NAME 'vkstream_verget_microbatch_500001_orafeed' NO COMMIT;

Rows Loaded
0

in the following link

https://my.vertica.com/docs/9.1.x/HTML/index.htm#Authoring/KafkaIntegrationGuide/UtilityOptions/LoadSpecUtilityOptions.htm

" --parser parser name

Identifies a Vertica UDParser to use with a specified target. This parser is used within the COPY statement that the scheduler runs to load data. If you are using a Vertica native parser, the values supplied to the --parser-parameters option are passed through to the COPY statement.
Default Value:
KafkaParser"

How to use native parser in LoadSpec???

anyone has anyidea?

What KafkaParser did is just treat your whole kafka message as an string and dump it into a String Column in Vertica table.
it is fine for FlexTable since it is schema-less and treat the whole string as a VMap object, but it didn't match the schema in your regular table.
If you setup your reject table in your microbatch, you should get some reject-reason hints from there. (you didn't specify it while you create your microbatch).

The solution should be:
since you are not using standard format like JSON or Avro but your own customized format, you should use delimiter to parse your message.
An example should be something like
vkconfig load-spec --create --config-schema verget --load-spec kafka --parser Delimited --parser-parameters "delimiter '|' "

Sorry we don't have an good example in our docs yet, we are working on to add them.

Thanks

Here's one more thing you probably also need pay attention on, only Kafka(Json/Avro)Parser is Kafka message awareness, which means know each message's size and can parse them properly. Then parsers or delimiter will only get a data stream cascade all the kafka message. To allow KafkaSource combine with other parsers, we need a Filter followed by the KafkaSource to add a temp record terminator between kafka messages, usually we use KafkaInsertDelimiters(E'\n') as the terminator.

See example here in the document:
https://my.vertica.com/docs/9.1.x/HTML/index.htm#Authoring/KafkaIntegrationGuide/ParsingCustomFormats.htm?TocPath=Integrating%20with%20Apache%20Kafka|_____8

Thanks a lot!! I will try it out and let you know.