Spark Streaming之foreachRDD操作详解

心已赠人 2022-12-29 02:20 201阅读 0赞

DStream中的所有计算，都是由output操作触发的，比如print()。如果没有任何output操作， 那么，压根儿就不会执行定义的计算逻辑。

此外，即使你使用了foreachRDD output操作，也必须在里面对RDD执行action操作，才能触 发对每一个batch的计算逻辑。否则，光有foreachRDD output操作，在里面没有对RDD执行 action操作，也不会触发任何逻辑。

<table> 
 <thead> 
  <tr> 
   <th>Output</th> 
   <th>Meaning</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>print</td> 
   <td>打印每个batch中的前10个元素，主要用于测试，或者是不需要执行什 么output操作时，用于简单触发一下job。</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFile(pref ix, [suffix])</td> 
   <td>将每个batch的数据保存到文件中。每个batch的文件的命名格式为： prefix-TIME_IN_MS[.suffix]</td> 
  </tr> 
  <tr> 
   <td>saveAsObjectFile</td> 
   <td>同上，但是将每个batch的数据以序列化对象的方式，保存到 SequenceFile中。</td> 
  </tr> 
  <tr> 
   <td>saveAsHadoopFile</td> 
   <td>同上，将数据保存到Hadoop文件中</td> 
  </tr> 
  <tr> 
   <td>foreachRDD</td> 
   <td>最常用的output操作，遍历DStream中的每个产生的RDD，进行处理。可 以将每个RDD中的数据写入外部存储，比如文件、数据库、缓存等。通常 在其中，是针对RDD执行action操作的，比如foreach。</td> 
  </tr> 
 </tbody> 
</table>

**通常在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();
    
        }
    }

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！