Перед отправкой Avro GenericRecord в Kafka заголовок вставляется следующим образом.
ProducerRecord<String, byte[]> record = new ProducerRecord<>(topicName, key, message);
record.headers().add("schema", schema);
Потребление записи.
При использовании Spark Streaming заголовок из ConsumerRecord остается нетронутым.
KafkaUtils.createDirectStream(streamingContext, LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, byte[]>Subscribe(topics, kafkaParams)).foreachRDD(rdd -> {
rdd.foreach(record -> {
System.out.println(new String(record.headers().headers("schema").iterator().next().value()));
});
});
;
Но при использовании Spark SQL Streaming заголовок отсутствует.
StreamingQuery query = dataset.writeStream().foreach(new ForeachWriter<>() {
...
@Override
public void process(Row row) {
String topic = (String) row.get(2);
int partition = (int) row.get(3);
long offset = (long) row.get(4);
String key = new String((byte[]) row.get(0));
byte[] value = (byte[]) row.get(1);
ConsumerRecord<String, byte[]> record = new ConsumerRecord<String, byte[]>(topic, partition, offset, key,
value);
//I need the schema to decode the Avro!
}
}).start();
Где я могу найти настраиваемое значение заголовка при использовании подхода Spark SQL Streaming?
Версия:
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.4.5</version>
ОБНОВЛЕНИЕ
Я попробовал 3.0.0-preview2 для spark-sql_2.12 и spark-sql-kafka-0-10_2.12. я добавил
.option("includeHeaders", true)
Но я по-прежнему получаю эти столбцы только из строки.
+---+-----+-----+---------+------+---------+-------------+
|key|value|topic|partition|offset|timestamp|timestampType|
+---+-----+-----+---------+------+---------+-------------+
