Sparkのあれらの事(二)Structured streamingの中でForeach sinkの使い方

Structured streamingのデフォルトでサポートされているsinkタイプは、File sink、Foreach sink、Console sink、Memory sinkです.特にForeach sinkの使い方を説明します(ps:Foreach sinkで外部redisに書き込む例).
lastEtlData.writeStream().foreach(new TestForeachWriter()).outputMode("update").start();
foreachメソッドのパラメータはForeachWriterオブジェクトです.apiの説明を参照してください.

datasetOfString.writeStream().foreach(new ForeachWriter() {
@Override
public boolean open(long partitionId, long version) {
  // open connection
  //        ， redis  ，   redis        
}

@Override
public void process(String value) {
  // write string to connection
  //        redis。value GenericRowWithSchema  
}

@Override
public void close(Throwable errorOrNull) {
  // close the connection
  //        
}
});

3つのメソッドの呼び出しプロセスを見てみましょう.各パーティのデータ呼び出しが1回なので、open、closeの頻度に注目する必要があります.一連のデータopen,closeはそれぞれ1回ずつ.
data.queryExecution.toRdd.foreachPartition { iter =>
if (writer.open(TaskContext.getPartitionId(), batchId)) {
try {
while (iter.hasNext) {
writer.process(encoder.fromRow(iter.next()))
}
} catch {
case e: Throwable =>
writer.close(e)
throw e
}
writer.close(null)
} else {
writer.close(null)
}
}
redisのforeachwriter実装を添付:

    public static class TestForeachWriter extends ForeachWriter implements Serializable{
   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(Object value) {
        GenericRowWithSchema genericRowWithSchema = (GenericRowWithSchema) value;

       System.out.println(((GenericRowWithSchema) value).get(0).toString()+"-----------"+ ((GenericRowWithSchema) value).get(2).toString());

        jedis.set(((GenericRowWithSchema) value).get(Integer.parseInt(genericRowWithSchema.schema().getFieldIndex("ID").get().toString())).toString(),((GenericRowWithSchema) value).get(Integer.parseInt(genericRowWithSchema.schema().getFieldIndex("COUNT(ADA1)").get().toString())).toString());
        System.out.println("++++++++++"+((GenericRowWithSchema) value).get(Integer.parseInt(genericRowWithSchema.schema().getFieldIndex("ID").get().toString())).toString());

    }

    @Override
    public void close(Throwable errorOrNull) {

        jedis.close();
    }
}

““
特に注意:Serializableインタフェースの実装を表示する必要があります.