DStream中的所有计算，都是由output操作触发的，比如print()。如果没有任何output操作， 那么，压根儿就不会执行定义的计算逻辑。
此外，即使你使用了foreachRDD output操作，也必须在里面对RDD执行action操作，才能触 发对每一个batch的计算逻辑。否则，光有foreachRDD output操作，在里面没有对RDD执行 action操作，也不会触发任何逻辑。
通常在foreachRDD中，都会创建一个Connection，比如JDBC Connection，然后通过Connection将数据写入外部存储。
误区一：在RDD的foreach操作外部，创建Connection
这种方式是错误的，因为它会导致Connection对象被序列化后传输到每个Task中。而这种Connection对象， 实际上一般是不支持序列化的，也就无法被传输。
dstream.foreachRDD( rdd => { val connection = createNewConnection() rdd.foreach( record => { connection.send(record) } })
误区二：在RDD的foreach操作内部，创建Connection
这种方式是可以的，但是效率低下。因为它会导致对于RDD中的每一条数据，都创建一个Connection对象。 而通常来说，Connection的创建，是很消耗性能的。
dstream.foreachRDD( rdd => { rdd.foreach( record => { val connection = createNewConnection() connection.send(record) connection.close() }) })
合理方式一：使用RDD的foreachPartition操作，并且在该操作内部，创建Connection对象
这样就相当于为RDD的每个partition创建一个Connection对象，节省资源的多了。
dstream.foreachRDD( rdd => { rdd.foreachPartition( partitionOfRecords => { val connection = createNewConnection() partitionOfRecords.foreach( record => { connection.send(record) connection.close() }) }) })
合理方式二：自己手动封装一个静态连接池，使用RDD的foreachPartition操作，并且在该操作内部，从 静态连接池中，通过静态方法，获取到一个连接，使用之后再还回去。
这样的话，甚至在多个RDD的 partition之间，也可以复用连接了。而且可以让连接池采取懒创建的策略，并且空闲一段时间后，将其释放掉。
dstream.foreachRDD( rdd => { rdd.foreachPartition( partitionOfRecords => { val connection = ConnectionPool.getConnection() partitionOfRecords.foreach( record => { connection.send(record) )} ConnectionPool.returnConnection(connection) }) })
案例：改写UpdateStateByKeyWordCount，将每次统计出来的全局的单词计数，写入一份，到MySQL数 据库中。
首先封装一个静态连接池：
package com.kfk.spark.common; import java.sql.Connection; import java.sql.DriverManager; import java.util.LinkedList; /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/18 * @time : 7:26 下午 */ public class ConnectionPool { private static LinkedList<Connection> connectionQueue; /** * 加载驱动 */ static { try { Class.forName("com.mysql.jdbc.Driver"); } catch (ClassNotFoundException e) { e.printStackTrace(); } } /** * 获取连接，多线程访问并发控制 * @return */ public synchronized static Connection getConnection(){ try { if (connectionQueue == null){ connectionQueue = new LinkedList<Connection>(); for (int i = 0;i < 10;i++){ Connection conn = DriverManager.getConnection( "jdbc:mysql://bigdata-pro-m04:3306/spark?useSSL=false", "root", "199911" ); connectionQueue.push(conn); } } } catch (Exception e){ e.printStackTrace(); } return connectionQueue.poll(); } /** * 还回去一个连接 * @param conn */ public static void returnConnection(Connection conn){ connectionQueue.push(conn); } }
编写实现功能代码：
package com.kfk.spark.common; import org.apache.spark.SparkConf; import org.apache.spark.streaming.Durations; import org.apache.spark.streaming.api.java.JavaStreamingContext; /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/14 * @time : 8:23 下午 */ public class CommStreamingContext { public static JavaStreamingContext getJssc(){ SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("CommStreamingContext"); return new JavaStreamingContext(conf, Durations.seconds(2)); } }
package com.kfk.spark.foreachrdd_project; import com.kfk.spark.common.CommStreamingContext; import com.kfk.spark.common.ConnectionPool; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.Optional; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.VoidFunction; import org.apache.spark.streaming.api.java.JavaDStream; import org.apache.spark.streaming.api.java.JavaPairDStream; import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; import org.apache.spark.streaming.api.java.JavaStreamingContext; import scala.Tuple2; import java.sql.Connection; import java.sql.Statement; import java.util.Arrays; import java.util.Iterator; import java.util.List; /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/18 * @time : 7:49 下午 */ public class ForeachPersistMySQL { public static void main(String[] args) throws InterruptedException { JavaStreamingContext jssc = CommStreamingContext.getJssc(); // 要使用UpdateStateByKey算子就必须设置一个Checkpoint目录，开启Checkpoint机制 // 以便于内存数据丢失时，可以从Checkpoint中恢复数据 jssc.checkpoint("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/sparkCheckpoint"); JavaReceiverInputDStream<String> lines = jssc.socketTextStream("bigdata-pro-m04",9999); // flatmap JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator()); // map JavaPairDStream<String,Integer> pair = words.mapToPair(word -> new Tuple2<>(word,1)); // 通过spark来维护一份每个单词的全局统计次数 JavaPairDStream<String,Integer> wordcount = pair.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { @Override public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception { Integer newValues = 0; if (state.isPresent()){ newValues = state.get(); } for (Integer value : values){ newValues += value; } return Optional.of(newValues); } }); // foreachRDD wordcount.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() { @Override public void call(JavaPairRDD<String, Integer> stringIntegerJavaPairRdd) throws Exception { stringIntegerJavaPairRdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() { @Override public void call(Iterator<Tuple2<String, Integer>> tuple2Iterator) throws Exception { Tuple2<String, Integer> wordcount = null; Connection conn = ConnectionPool.getConnection(); while (tuple2Iterator.hasNext()){ wordcount = tuple2Iterator.next(); String sql = "insert into spark.wordcount(word,count) values('"+wordcount._1+"', '"+wordcount._2+"')"; Statement statement = conn.createStatement(); statement.executeUpdate(sql); } ConnectionPool.returnConnection(conn); } }); } }); jssc.start(); jssc.awaitTermination(); jssc.close(); } }
