Read protobuf kafka message using spark structured streaming

Question

Is it possible to read protobuf message from kafka using spark structured streaming?

Niket Arora · Accepted Answer · 2018-08-23 16:46:56Z

2

Approach 1

sparkSession.udf().register("deserialize", getDeserializer(), schema);

    DataStreamReader dataStreamReader = sparkSession.readStream().format("kafka");

    for (Map.Entry<String, String> kafkaPropEntry : kafkaProps.entrySet()) {
        dataStreamReader.option(kafkaPropEntry.getKey(), kafkaPropEntry.getValue());
    }

    Dataset<Row> kafkaRecords =
            dataStreamReader.load()
                    .selectExpr("deserialize(value) as event").select("event.*");

Approach 2

final StructType schema = getSchema();

    DataStreamReader dataStreamReader = sparkSession.readStream().format("kafka");

    for (Map.Entry<String, String> kafkaPropEntry : kafkaProps.entrySet()) {
        dataStreamReader.option(kafkaPropEntry.getKey(), kafkaPropEntry.getValue());
    }

    Dataset<Row> kafkaRecords = dataStreamReader.load()
            .map(row -> getOutputRow((byte[]) row.get(VALUE_INDEX)), RowEncoder.apply(schema))

Approach 1 has one flaw as deserialize method is called multiple times (for evert column in event) https://issues.apache.org/jira/browse/SPARK-17728. Approach 2 maps protobuf to row directly using map method.

answered Aug 23, 2018 at 16:46

Niket Arora

711 silver badge5 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Steephen Over a year ago

What is this getOutputRow()?

Collectives™ on Stack Overflow

Read protobuf kafka message using spark structured streaming

1 Answer 1

Approach 1

Approach 2

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Approach 1

Approach 2

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related