FCSVPARSER in Kafka

ChiaraB

Hello,
I have a csv file (with 57 fields) that I sent to kafka. Now I would like to load these data from kafka in a Vertica flex table using the FCSVPARSER parser with the command:
$ vkconfig load-spec --create --config-schema PDMScheduler --load-spec PDM_load_spec --parser FCSVPARSER

All seems correctly configured but the table is not loaded and no records in the rejected table .
If I used the copy command with the same parser to load data in the flex table and is OK.
Where am I wrong ? Is the use of FCSVPARSER allowed in vkconfig?

This is what I see in vertica.log file

2017-07-24 15:16:20.514 Init Session:7f1ea6079700-a0000000d15508 @v_haven_node0001: 00000/3316: Executing statement: 'COPY "pdm"."signal_data_flex" SOURCE KafkaSource(stream='pointnext1|0|32601', brokers='10.3.150.60:9092', duration=interval '9876 milliseconds', eof_timeout=interval '1000 milliseconds', stop_on_eof=true, message_max_bytes=1048576 ) PARSER FCSVPARSER() REJECTED DATA AS TABLE "PDM"."signal_data_flex_rej" TRICKLE STREAM NAME 'vkstream_PDMScheduler_microbatch_1_pdm_signal' NO COMMIT'
2017-07-24 15:16:20.518 Init Session:7f1ea6079700-a0000000d15508 [LocalPlanner] finalizeThreadsToUseForLoad(): max_threads_allowed = 24, targetThreads = 1
2017-07-24 15:16:20.519 Init Session:7f1ea6079700-a0000000d15508 [LocalPlanner] finalizeThreadsToUseForLoad(): cannot apportion, cannot cooperate
2017-07-24 15:16:20.522 EEcmdq:7f1e0ffff700 [UserMessage] COPY - invalid value (1000000) for kafka setting metadata.request.timeout.ms: Configuration property "metadata.request.timeout.ms" value 1000000 is outside allowed range 10..900000

2017-07-24 15:16:20.522 EEcmdq:7f1e0ffff700 [UserMessage] COPY - invalid value (1000000) for kafka setting socket.timeout.ms: Configuration property "socket.timeout.ms" value 1000000 is outside allowed range 10..300000

Thanks

Chiara

sKwa

Hi.

> invalid value (1000000) for kafka setting socket.timeout.ms: Configuration property "socket.timeout.ms" value 1000000 is outside allowed range 10..300000

????

ChiaraB

I saw the invalid values, but kafka streaming on a standard table with DELIMITED parser works even with these messages in the log file.
However have these configuration setting to be modified in the kafka server ?

ChiaraB

Another step debugging was to try manually stream data from Kafka using a COPY statement.
The following statement (that use the fcsvparser ) didn't load anything (no record also in the rej table) and no errors are shown .
COPY pdm.signal_data_flex
SOURCE KafkaSource( stream='pointnext1|0|-2', brokers='10.3.150.60:9092',duration=interval '10000 milliseconds',executionparallelism='1',stop_on_eof=true)
PARSER fcsvparser()
REJECTED DATA AS TABLE signal_data_flex_rej1
DIRECT;

Rows Loaded

       0

(1 row)
while the same statement without a parser load just a single row:
COPY pdm.signal_data_flex
SOURCE KafkaSource( stream='pointnext1|0|-2', brokers='10.3.150.60:9092',duration=interval '10000 milliseconds',executionparallelism='1',stop_on_eof=true)
REJECTED DATA AS TABLE signal_data_flex_rej2
DIRECT;

Rows Loaded

       1

(1 row)

It seems tha fscvparser is not allowed but however no message errors are shown.
Someone could confirm this ?

Thanks
Chiara