Как создать преобразование в Memsql, если источником является формат Kafka Avro

Я могу передавать данные из Kafka в Memsql.

Я пытаюсь нажать с помощью Transform. Я создал Kafka Consumer на Python, который использует данные из Kafka Topic и конвертирует их в формат Json.

Я не знаю, как использовать это как Transform в Memsql.

from confluent_kafka import KafkaError
from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError
import sys

c = AvroConsumer({
    'bootstrap.servers': 'X.Y.Z.W:9092',
    'group.id': 'groupid1112',
    'schema.registry.url': 'http://X.Y.Z.W:8081',
    'default.topic.config': {
        'auto.offset.reset': 'smallest'
    }
    })

c.subscribe(['test_topic'])
count =0
while True:
    try:
        msg = c.poll(10)

    except SerializerError as e:
        print("Message deserialization failed for {}: {}".format(msg, e))
        break

    if msg is None:
        continue

    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break
    valueList = list(msg.value().values())
    print(valueList)

c.close()

это печать

[1518776144187, 1, 2, 103,'asas',asas'eer',None]

python apache-kafka confluent-schema-registry confluent-platform singlestore
person cody123    schedule 04.08.2018    source источник
comment
попробуйте print(msg.value()), иначе у MemSQL есть драйвер JDBC, вы можете использовать с JDBC Kafka Connect для чтения данных Avro и записи в MemSQL   -  person OneCricketeer    schedule 01.09.2018

Ответы (1)


arrow_upward
0
arrow_downward

проверьте эти документы https://docs.memsql.com/memsql-pipelines/v6.0/transforms/

следите за встроенной поддержкой avro в следующем выпуске MemSQL.

вы захотите сделать что-то вроде следующего, но со мной, набрасывающим детали, специфичные для avro, так как я не знаю библиотеку avro навскидку.

```

def input_stream():
    """
        Consume STDIN and yield each record that is received from MemSQL
    """
    while True:
        byte_len = sys.stdin.read(8)
        if len(byte_len) == 8:
            byte_len = struct.unpack("L", byte_len)[0]
            result = sys.stdin.read(byte_len)
            yield result
        else:
            assert len(byte_len) == 0, byte_len
            return

avro_context = WhateverYouNeed() # maybe connect to schema registry here if you need to

for msg in input_stream():
    object = DeserializeAvro(avro_context, msg) # this is your code
    sys.stdout.write(SerializeToTSV(object)) # also your code

```

Использование реестра схем должно быть в порядке, но вам не нужно беспокоиться о деталях чтения из kafka в вашем сценарии преобразования. Я могу попытаться предоставить вам более подробный сценарий в понедельник, но это то, как структурировать код.

person Joseph Victor    schedule 04.08.2018
comment
Я просмотрел документацию, но нет возможности настроить реестр в конвейере, поэтому я только спросил, как использовать преобразование в случае avro. - person cody123; 05.08.2018