Kafka：ZK+Kafka+Spark Streaming集群环境搭建（十六）Structured Streaming中ForeachSink的用法

Structured Streaming默认支持的sink类型有File sink，Foreach sink，Console sink，Memory sink。

ForeachWriter实现：

以写入redis为例

package com.dx.streaming.producer;

import org.apache.spark.sql.ForeachWriter;
import org.apache.spark.sql.Row;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class TestForeachWriter extends ForeachWriter<Row> {
    private static final long serialVersionUID = 1801843595306161029L;

    public static JedisPool jedisPool;
    public Jedis jedis;
    static {
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(20);
            config.setMaxIdle(5);
            config.setMaxWaitMillis(1000);
            config.setMinIdle(2);
            config.setTestOnBorrow(false);
            jedisPool = new JedisPool(config, "127.0.0.1", 6379);
    }

    public static synchronized Jedis getJedis() {
        return jedisPool.getResource();
    }


    @Override
    public boolean open(long partitionId, long version) {
           jedis = getJedis();
           return true;
    }

    @Override
    public void process(Row row) {
         jedis.set("row.key","row.value");
    }

    @Override
    public void close(Throwable arg0) {
         jedis.close();
    }
}

Structured Streaming中使用ForeachWriter示例：

package com.dx.streaming.producer;

import java.util.HashMap;
import java.util.Map;

import org.apache.avro.Schema;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

import com.databricks.spark.avro.SchemaConverters;

public class TestConsumer {
    private static final String avroFilePath = "D:\Java_Study\workspace\kafka-streaming-learn\conf\avro\userlog.avsc";
    //private static final String avroFilePath = "/user/dx/conf/avro/userlog.avsc";
    private static final String topic = "t-my";

    public static void main(String[] args) throws Exception {
        String appName = "Test Avro";
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName(appName);
        SparkSession sparkSession = SparkSession.builder().config(conf).getOrCreate();

        Map<String, String> kafkaOptions = new HashMap<String, String>();
        kafkaOptions.put("kafka.bootstrap.servers", "192.168.0.121:9092");
        
        Schema schema = SchemaUtil.getAvroSchemaFromHDFSFile(avroFilePath);    
        AvroParserUDF udf = new AvroParserUDF(avroFilePath);
        StructType type = (StructType) SchemaConverters.toSqlType(schema).dataType();        
        sparkSession.udf().register("deserialize", udf, DataTypes.createStructType(type.fields()));

        Dataset<Row> stream = sparkSession.readStream().format("kafka").options(kafkaOptions).option("subscribe", topic).option("startingOffsets", "earliest").load().select("value").as(Encoders.BINARY())
                .selectExpr("deserialize(value) as row").select("row.*");

        stream.printSchema();

        // Print new data to console
        StreamingQuery query = stream.writeStream().foreach(new TestForeachWriter()).outputMode("update").start();
        
        try {
            query.awaitTermination();
            sparkSession.streams().awaitAnyTermination();
        } catch (StreamingQueryException e) {
            e.printStackTrace();
        }
    }
}

参考：Spark的那些事(二)Structured streaming中Foreach sink的用法