Чтение разделенных трубами значений в ksql - PullRequest
Новая
       38

Чтение разделенных трубами значений в ksql

0 голосов
/ 11 марта 2020

Я работаю над PO C, мне нужно прочитать файл значений, разделенных конвейерами, и вставить эти записи в сервер ms sql. Я использую confluent 5.4.1 для использования value_delimiter create stream property. Но это дает исключение: Delimeter only supported with DELIMITED format

1. Start Confluent (версия: 5.4.1) :: 

[Dev root @ myip ~]
# confluent local start
    The local commands are intended for a single-node development environment
    only, NOT for production usage. https://docs.confluent.io/current/cli/index.html

Using CONFLUENT_CURRENT: /tmp/confluent.vHhSRAnj
Starting zookeeper
zookeeper is [UP]
Starting kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting kafka-rest
kafka-rest is [UP]
Starting connect
connect is [UP]
Starting ksql-server
ksql-server is [UP]
Starting control-center
control-center is [UP]
[Dev root @ myip ~]
# jps
49923 KafkaRestMain
50099 ConnectDistributed
49301 QuorumPeerMain
50805 KsqlServerMain
49414 SupportedKafka
52103 Jps
51020 ControlCenter
1741
49646 SchemaRegistryMain
[Dev root @ myip ~]
#

2. Создать топи c: 

[Dev root @ myip ~]
# kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic SampleData
Created topic SampleData.

3. Предоставьте разделенные по конвейеру данные для SampeData Topi c

[Dev root @ myip ~]
# kafka-console-producer --broker-list localhost:9092 --topic SampleData <<EOF
> this is col1|and now col2|and col 3 :)
> EOF
>>[Dev root @ myip ~]
#

4. Старт К SQL :: 

[Dev root @ myip ~]
# ksql

                  ===========================================
                  =        _  __ _____  ____  _             =
                  =       | |/ // ____|/ __ \| |            =
                  =       | ' /| (___ | |  | | |            =
                  =       |  <  \___ \| |  | | |            =
                  =       | . \ ____) | |__| | |____        =
                  =       |_|\_\_____/ \___\_\______|       =
                  =                                         =
                  =  Streaming SQL Engine for Apache Kafka® =
                  ===========================================

Copyright 2017-2019 Confluent Inc.

CLI v5.4.1, Server v5.4.1 located at http://localhost:8088

Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!

5. Объявите схему для существующего topi c: SampleData 

ksql> CREATE STREAM sample_delimited (
>       column1 varchar(1000),
>       column2 varchar(1000),
>       column3 varchar(1000))
>       WITH (KAFKA_TOPIC='SampleData', VALUE_FORMAT='DELIMITED', VALUE_DELIMITER='|');

 Message
----------------
 Stream created
----------------

6. Проверьте данные в K SQl Stream 


ksql>  SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
ksql> SELECT * FROM sample_delimited emit changes limit 1;
+---------------------------+---------------------------+---------------------------+---------------------------+---------------------------+
|ROWTIME                    |ROWKEY                     |COLUMN1                    |COLUMN2                    |COLUMN3                    |
+---------------------------+---------------------------+---------------------------+---------------------------+---------------------------+
|1584339233947              |null                       |this is col1               |and now col2               |and col 3 :)               |
Limit Reached
Query terminated

7. Напишите новую Kafka topi c: SampleDataAvro, которая сериализует все данные из потока sample_delimited в поток формата Avro 

ksql> CREATE STREAM sample_avro WITH (KAFKA_TOPIC='SampleDataAvro', VALUE_FORMAT='AVRO') AS SELECT * FROM sample_delimited;
Delimeter only supported with DELIMITED format
ksql>

8. Выше строки дает исключение ::

Delimeter only supported with DELIMITED format

9. Нагрузка мс sql конфигурация подключения кафки

confluent local load test-sink -- -d ./etc/kafka-connect-jdbc/sink-quickstart-mssql.properties

1 Ответ

0 голосов
/ 11 марта 2020

Единственный раз, когда вам нужно указать разделитель, это когда вы определяете поток, который читает из источника topi c.

Вот мой рабочий пример:

  1. Заполните topi c данными с разделителями в виде трубы: 

    $ kafkacat -b localhost:9092 -t SampleData -P<<EOF
this is col1|and now col2|and col 3 :)
EOF

  2. Объявите поток поверх него 

    CREATE STREAM sample_delimited ( 
        column1 varchar(1000),
        column2 varchar(1000),
        column3 varchar(1000)) 
        WITH (KAFKA_TOPIC='SampleData', VALUE_FORMAT='DELIMITED', VALUE_DELIMITER='|');

  3. Запросите поток, чтобы убедиться, что он работает 

    ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.

ksql> SELECT * FROM sample_delimited emit changes limit 1;
+----------------+--------+---------------+--------------+--------------+
|ROWTIME         |ROWKEY  |COLUMN1        |COLUMN2       |COLUMN3       |
+----------------+--------+---------------+--------------+--------------+
|1583933584658   |null    |this is col1   |and now col2  |and col 3 :)  |
Limit Reached
Query terminated

  4. Повторно отправьте данные в Avro: 

    CREATE STREAM sample_avro WITH (KAFKA_TOPIC='SampleDataAvro', VALUE_FORMAT='AVRO') AS SELECT * FROM sample_delimited;

  5. Дамп содержимого топи c - обратите внимание, что сейчас Avro: 

    ksql> print SampleDataAvro;
Key format: UNDEFINED
Value format: AVRO
rowtime: 3/11/20 1:33:04 PM UTC, key: <null>, value: {"COLUMN1": "this is col1", "COLUMN2": "and now col2", "COLUMN3": "and col 3 :)"}

ошибка, которую вы попали в результате ошибки # 4200 . Вы можете дождаться следующего выпуска Confluent Platform или использовать standalone ksqlDB , в котором проблема уже устранена.

Здесь используется ksqlDB 0.7.1 для потоковой передачи данных в MS SQL: 

CREATE SINK CONNECTOR SINK_MSSQL WITH (
    'connector.class'     = 'io.confluent.connect.jdbc.JdbcSinkConnector',
    'connection.url'      = 'jdbc:sqlserver://mssql:1433',
    'connection.user'     = 'sa',
    'connection.password' = 'Admin123',
    'topics'              = 'SampleDataAvro',
    'key.converter'       = 'org.apache.kafka.connect.storage.StringConverter',
    'auto.create'         = 'true',
    'insert.mode'         = 'insert'
  );

Теперь запросите данные в MS SQL

1> Select @@version
2> go

---------------------------------------------------------------------
Microsoft SQL Server 2017 (RTM-CU17) (KB4515579) - 14.0.3238.1 (X64)
        Sep 13 2019 15:49:57
        Copyright (C) 2017 Microsoft Corporation
        Developer Edition (64-bit) on Linux (Ubuntu 16.04.6 LTS)

(1 rows affected)

1> SELECT * FROM SampleDataAvro;
2> GO
COLUMN3        COLUMN2         COLUMN1     
-------------- --------------- ------------------
and col 3 :)   and now col2    this is col1

(1 rows affected)
...