Skip to main content

FCSVPARSER in Kafka

  • July 24, 2017
  • 3 replies
  • 20 views
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

    3 replies

    sKwa
    Forum|alt.badge.img+1
    • sKwa
    • Participating Frequently
    • July 24, 2017
    Hi.

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

    ????
    ChiaraB
    • ChiaraB
    • Author
    • New Participant
    • July 24, 2017

    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
    • ChiaraB
    • Author
    • New Participant
    • July 25, 2017

    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

    Still scrolling?
    Ask a question
    Terms & ConditionsAccessibility statement