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

How To Use

Kafka uses example

  • This is an example to generate data from fake source and sink to kafka with avro format.
  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   FakeSource {
  8.     row.num = 90
  9.     schema = {
  10.       fields {
  11.         c_map = "map<string, string>"
  12.         c_array = "array<int>"
  13.         c_string = string
  14.         c_boolean = boolean
  15.         c_tinyint = tinyint
  16.         c_smallint = smallint
  17.         c_int = int
  18.         c_bigint = bigint
  19.         c_float = float
  20.         c_double = double
  21.         c_bytes = bytes
  22.         c_date = date
  23.         c_decimal = "decimal(38, 18)"
  24.         c_timestamp = timestamp
  25.         c_row = {
  26.           c_map = "map<string, string>"
  27.           c_array = "array<int>"
  28.           c_string = string
  29.           c_boolean = boolean
  30.           c_tinyint = tinyint
  31.           c_smallint = smallint
  32.           c_int = int
  33.           c_bigint = bigint
  34.           c_float = float
  35.           c_double = double
  36.           c_bytes = bytes
  37.           c_date = date
  38.           c_decimal = "decimal(38, 18)"
  39.           c_timestamp = timestamp
  40.         }
  41.       }
  42.     }
  43.     result_table_name = "fake"
  44.   }
  45. }
  47. sink {
  48.   Kafka {
  49.     bootstrap.servers = "kafkaCluster:9092"
  50.     topic = "test_avro_topic_fake_source"
  51.     format = avro
  52.   }
  53. }
  • This is an example read data from kafka with avro format and print to console.
  1. env {
  2.   parallelism = 1
  3.   job.mode = "BATCH"
  4. }
  6. source {
  7.   Kafka {
  8.     bootstrap.servers = "kafkaCluster:9092"
  9.     topic = "test_avro_topic"
  10.     result_table_name = "kafka_table"
  11.     kafka.auto.offset.reset = "earliest"
  12.     format = avro
  13.     format_error_handle_way = skip
  14.     schema = {
  15.       fields {
  16.         id = bigint
  17.         c_map = "map<string, smallint>"
  18.         c_array = "array<tinyint>"
  19.         c_string = string
  20.         c_boolean = boolean
  21.         c_tinyint = tinyint
  22.         c_smallint = smallint
  23.         c_int = int
  24.         c_bigint = bigint
  25.         c_float = float
  26.         c_double = double
  27.         c_decimal = "decimal(2, 1)"
  28.         c_bytes = bytes
  29.         c_date = date
  30.         c_timestamp = timestamp
  31.       }
  32.     }
  33.   }
  34. }
  36. sink {
  37.   Console {
  38.     source_table_name = "kafka_table"
  39.   }
  40. }