Skip to content

[Improve] improve avro format convert #6082

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Hisoka-X merged 4 commits into from
Jan 5, 2024
Merged

[Improve] improve avro format convert #6082

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2b93dfd
[Improve] improve avro format convert
liunaijie Dec 25, 2023
ac6e2d1
[Improve] improve kafka avro e2e
liunaijie Dec 28, 2023
b891202
[Feature] add avro format document
liunaijie Jan 4, 2024
7235aed
[Improve] update avro bytes type convert
liunaijie Jan 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 111 additions & 0 deletions docs/en/connector-v2/formats/avro.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
# Avro format

Avro is very popular in streaming data pipeline. Now seatunnel supports Avro format in kafka connector.

# How to use Avro format

## Kafka uses example

- This is an example to generate data from fake source and sink to kafka with avro format.

```bash
env {
parallelism = 1
job.mode = "BATCH"
}

source {
FakeSource {
row.num = 90
schema = {
fields {
c_map = "map<string, string>"
c_array = "array<int>"
c_string = string
c_boolean = boolean
c_tinyint = tinyint
c_smallint = smallint
c_int = int
c_bigint = bigint
c_float = float
c_double = double
c_bytes = bytes
c_date = date
c_decimal = "decimal(38, 18)"
c_timestamp = timestamp
c_row = {
c_map = "map<string, string>"
c_array = "array<int>"
c_string = string
c_boolean = boolean
c_tinyint = tinyint
c_smallint = smallint
c_int = int
c_bigint = bigint
c_float = float
c_double = double
c_bytes = bytes
c_date = date
c_decimal = "decimal(38, 18)"
c_timestamp = timestamp
}
}
}
result_table_name = "fake"
}
}

sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
topic = "test_avro_topic_fake_source"
format = avro
}
}
```

- This is an example read data from kafka with avro format and print to console.

```bash
env {
parallelism = 1
job.mode = "BATCH"
}

source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
topic = "test_avro_topic"
result_table_name = "kafka_table"
kafka.auto.offset.reset = "earliest"
format = avro
format_error_handle_way = skip
schema = {
fields {
id = bigint
c_map = "map<string, smallint>"
c_array = "array<tinyint>"
c_string = string
c_boolean = boolean
c_tinyint = tinyint
c_smallint = smallint
c_int = int
c_bigint = bigint
c_float = float
c_double = double
c_decimal = "decimal(2, 1)"
c_bytes = bytes
c_date = date
c_timestamp = timestamp
}
}
}
}

sink {
Console {
source_table_name = "kafka_table"
}
}
```

183 changes: 176 additions & 7 deletions ...e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.seatunnel.e2e.common.container.TestContainerId;
import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
import org.apache.seatunnel.format.text.TextSerializationSchema;

import org.apache.kafka.clients.consumer.ConsumerConfig;
Expand All @@ -47,6 +48,7 @@
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringDeserializer;

Expand All @@ -66,6 +68,7 @@

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
Expand Down Expand Up @@ -294,27 +297,150 @@ public void testSourceKafkaStartConfig(TestContainer container)
}

@TestTemplate
@DisabledOnContainer(TestContainerId.SPARK_2_4)
@DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
public void testFakeSourceToKafkaAvroFormat(TestContainer container)
throws IOException, InterruptedException {
Container.ExecResult execResult =
container.executeJob("/avro/fake_source_to_kafka_avro_format.conf");
Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
String[] subField = {
"c_map",
"c_array",
"c_string",
"c_boolean",
"c_tinyint",
"c_smallint",
"c_int",
"c_bigint",
"c_float",
"c_double",
"c_bytes",
"c_date",
"c_decimal",
"c_timestamp"
};
SeaTunnelDataType<?>[] subFieldTypes = {
new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE),
ArrayType.INT_ARRAY_TYPE,
BasicType.STRING_TYPE,
BasicType.BOOLEAN_TYPE,
BasicType.BYTE_TYPE,
BasicType.SHORT_TYPE,
BasicType.INT_TYPE,
BasicType.LONG_TYPE,
BasicType.FLOAT_TYPE,
BasicType.DOUBLE_TYPE,
PrimitiveByteArrayType.INSTANCE,
LocalTimeType.LOCAL_DATE_TYPE,
new DecimalType(38, 18),
LocalTimeType.LOCAL_DATE_TIME_TYPE
};
SeaTunnelRowType subRow = new SeaTunnelRowType(subField, subFieldTypes);
String[] fieldNames = {
"c_map",
"c_array",
"c_string",
"c_boolean",
"c_tinyint",
"c_smallint",
"c_int",
"c_bigint",
"c_float",
"c_double",
"c_bytes",
"c_date",
"c_decimal",
"c_timestamp",
"c_row"
};
SeaTunnelDataType<?>[] fieldTypes = {
new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE),
ArrayType.INT_ARRAY_TYPE,
BasicType.STRING_TYPE,
BasicType.BOOLEAN_TYPE,
BasicType.BYTE_TYPE,
BasicType.SHORT_TYPE,
BasicType.INT_TYPE,
BasicType.LONG_TYPE,
BasicType.FLOAT_TYPE,
BasicType.DOUBLE_TYPE,
PrimitiveByteArrayType.INSTANCE,
LocalTimeType.LOCAL_DATE_TYPE,
new DecimalType(38, 18),
LocalTimeType.LOCAL_DATE_TIME_TYPE,
subRow
};
SeaTunnelRowType fake_source_row_type = new SeaTunnelRowType(fieldNames, fieldTypes);
AvroDeserializationSchema avroDeserializationSchema =
new AvroDeserializationSchema(fake_source_row_type);
List<SeaTunnelRow> kafkaSTRow =
getKafkaSTRow(
"test_avro_topic_fake_source",
value -> {
try {
return avroDeserializationSchema.deserialize(value);
} catch (IOException e) {
throw new RuntimeException(e);
}
});
Assertions.assertEquals(90, kafkaSTRow.size());
kafkaSTRow.forEach(
row -> Assertions.assertEquals("fake_source_avro", row.getField(2).toString()));
Comment thread
Hisoka-X marked this conversation as resolved.
Outdated
}

@TestTemplate
@DisabledOnContainer(TestContainerId.SPARK_2_4)
public void testKafkaAvroToConsole(TestContainer container)
@DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
public void testKafkaAvroToAssert(TestContainer container)
throws IOException, InterruptedException {
DefaultSeaTunnelRowSerializer serializer =
DefaultSeaTunnelRowSerializer.create(
"test_avro_topic",
SEATUNNEL_ROW_TYPE,
MessageFormat.AVRO,
DEFAULT_FIELD_DELIMITER);
generateTestData(row -> serializer.serializeRow(row), 0, 100);
Container.ExecResult execResult = container.executeJob("/avro/kafka_avro_to_console.conf");
int start = 0;
int end = 100;
generateTestData(row -> serializer.serializeRow(row), start, end);
Container.ExecResult execResult = container.executeJob("/avro/kafka_avro_to_assert.conf");
Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());

AvroDeserializationSchema avroDeserializationSchema =
new AvroDeserializationSchema(SEATUNNEL_ROW_TYPE);
List<SeaTunnelRow> kafkaSTRow =
getKafkaSTRow(
"test_avro_topic",
value -> {
try {
return avroDeserializationSchema.deserialize(value);
} catch (IOException e) {
throw new RuntimeException(e);
}
});
Assertions.assertEquals(100, kafkaSTRow.size());
kafkaSTRow.forEach(
row -> {
Assertions.assertTrue(
(long) row.getField(0) >= start && (long) row.getField(0) < end);
Assertions.assertEquals(
Collections.singletonMap("key", Short.parseShort("1")),
(Map<String, Short>) row.getField(1));
Assertions.assertArrayEquals(
new Byte[] {Byte.parseByte("1")}, (Byte[]) row.getField(2));
Assertions.assertEquals("string", row.getField(3).toString());
Assertions.assertEquals(false, row.getField(4));
Assertions.assertEquals(Byte.parseByte("1"), row.getField(5));
Assertions.assertEquals(Short.parseShort("1"), row.getField(6));
Assertions.assertEquals(Integer.parseInt("1"), row.getField(7));
Assertions.assertEquals(Long.parseLong("1"), row.getField(8));
Assertions.assertEquals(Float.parseFloat("1.1"), row.getField(9));
Assertions.assertEquals(Double.parseDouble("1.1"), row.getField(10));
Assertions.assertEquals(BigDecimal.valueOf(11, 1), row.getField(11));
Assertions.assertArrayEquals(
"test".getBytes(), ((ByteBuffer) row.getField(12)).array());
Assertions.assertEquals(LocalDate.of(2024, 1, 1), row.getField(13));
Assertions.assertEquals(
LocalDateTime.of(2024, 1, 1, 12, 59, 23), row.getField(14));
});
}

public void testKafkaLatestToConsole(TestContainer container)
Expand Down Expand Up @@ -373,6 +499,22 @@ private Properties kafkaConsumerConfig() {
return props;
}

private Properties kafkaByteConsumerConfig() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "seatunnel-kafka-sink-group");
props.put(
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
OffsetResetStrategy.EARLIEST.toString().toLowerCase());
props.setProperty(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
props.setProperty(
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class.getName());
return props;
}

private void generateTestData(ProducerRecordConverter converter, int start, int end) {
for (int i = start; i < end; i++) {
SeaTunnelRow row =
Expand All @@ -391,8 +533,8 @@ private void generateTestData(ProducerRecordConverter converter, int start, int
Double.parseDouble("1.1"),
BigDecimal.valueOf(11, 1),
"test".getBytes(),
LocalDate.now(),
LocalDateTime.now()
LocalDate.of(2024, 1, 1),
LocalDateTime.of(2024, 1, 1, 12, 59, 23)
});
ProducerRecord<byte[], byte[]> producerRecord = converter.convert(row);
producer.send(producerRecord);
Expand Down Expand Up @@ -480,7 +622,34 @@ private List<String> getKafkaConsumerListData(String topicName) {
return data;
}

private List<SeaTunnelRow> getKafkaSTRow(String topicName, ConsumerRecordConverter converter) {
List<SeaTunnelRow> data = new ArrayList<>();
try (KafkaConsumer<byte[], byte[]> consumer =
new KafkaConsumer<>(kafkaByteConsumerConfig())) {
consumer.subscribe(Arrays.asList(topicName));
Map<TopicPartition, Long> offsets =
consumer.endOffsets(Arrays.asList(new TopicPartition(topicName, 0)));
Long endOffset = offsets.entrySet().iterator().next().getValue();
Long lastProcessedOffset = -1L;

do {
ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<byte[], byte[]> record : records) {
if (lastProcessedOffset < record.offset()) {
data.add(converter.convert(record.value()));
}
lastProcessedOffset = record.offset();
}
} while (lastProcessedOffset < endOffset - 1);
}
return data;
}

interface ProducerRecordConverter {
ProducerRecord<byte[], byte[]> convert(SeaTunnelRow row);
}

interface ConsumerRecordConverter {
SeaTunnelRow convert(byte[] value);
}
}
4 changes: 3 additions & 1 deletion ...-v2-e2e/connector-kafka-e2e/src/test/resources/avro/fake_source_to_kafka_avro_format.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ env {

source {
FakeSource {
row.num = 90
string.template = ["fake_source_avro"]
schema = {
fields {
c_map = "map<string, string>"
Expand Down Expand Up @@ -70,7 +72,7 @@ source {
sink {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
topic = "test_avro_topic"
topic = "test_avro_topic_fake_source"
format = avro
}
}
Loading