一、output操作
1、output操作
DStream中的所有計算，都是由output操作觸發的，比如print()。如果沒有任何output操作，那么，壓根兒就不會執行定義的計算邏輯。
此外，即使你使用了foreachRDD output操作，也必須在里面對RDD執行action操作，才能觸發對每一個batch的計算邏輯。否則，光有
foreachRDD output操作，在里面沒有對RDD執行action操作，也不會觸發任何邏輯。
2、output操作概覽
二、foreachRDD
1、foreachRDD詳解
通常在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數據庫中。
2、java案例
創建mysql表
mysql> use testdb; Reading table information for completion of table and column names You can turn off this feature to get a quicker startup with -A Database changed mysql> create table wordcount ( -> id integer auto_increment primary key, -> updated_time timestamp NOT NULL default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP, -> word varchar(255), -> count integer -> ); Query OK, 0 rows affected (0.05 sec)
java代碼
###ConnectionPool package cn.spark.study.streaming; import java.sql.Connection; import java.sql.DriverManager; import java.util.LinkedList; /** * 簡易版的連接池 * @author Administrator * */ public class ConnectionPool { // 靜態的Connection隊列 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://spark1:3306/testdb", "", ""); connectionQueue.push(conn); } } } catch (Exception e) { e.printStackTrace(); } return connectionQueue.poll(); } /** * 還回去一個連接 */ public static void returnConnection(Connection conn) { connectionQueue.push(conn); } } ###PersistWordCount package cn.spark.study.streaming; import java.sql.Connection; import java.sql.Statement; import java.util.Arrays; import java.util.Iterator; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.api.java.function.VoidFunction; import org.apache.spark.streaming.Durations; 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 com.google.common.base.Optional; import scala.Tuple2; /** * 基於持久化機制的實時wordcount程序 * @author Administrator * */ public class PersistWordCount { public static void main(String[] args) { SparkConf conf = new SparkConf() .setMaster("local[2]") .setAppName("PersistWordCount"); JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5)); jssc.checkpoint("hdfs://spark1:9000/wordcount_checkpoint"); JavaReceiverInputDStream<String> lines = jssc.socketTextStream("spark1", 9999); JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { private static final long serialVersionUID = 1L; @Override public Iterable<String> call(String line) throws Exception { return Arrays.asList(line.split(" ")); } }); JavaPairDStream<String, Integer> pairs = words.mapToPair( new PairFunction<String, String, Integer>() { private static final long serialVersionUID = 1L; @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); JavaPairDStream<String, Integer> wordCounts = pairs.updateStateByKey( new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() { private static final long serialVersionUID = 1L; @Override public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception { Integer newValue = 0; if(state.isPresent()) { newValue = state.get(); } for(Integer value : values) { newValue += value; } return Optional.of(newValue); } }); // 每次得到當前所有單詞的統計次數之后，將其寫入mysql存儲，進行持久化，以便於后續的J2EE應用程序 // 進行顯示 wordCounts.foreachRDD(new Function<JavaPairRDD<String,Integer>, Void>() { private static final long serialVersionUID = 1L; @Override public Void call(JavaPairRDD<String, Integer> wordCountsRDD) throws Exception { // 調用RDD的foreachPartition()方法 wordCountsRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String,Integer>>>() { private static final long serialVersionUID = 1L; @Override public void call(Iterator<Tuple2<String, Integer>> wordCounts) throws Exception { // 給每個partition，獲取一個連接 Connection conn = ConnectionPool.getConnection(); // 遍歷partition中的數據，使用一個連接，插入數據庫 Tuple2<String, Integer> wordCount = null; while(wordCounts.hasNext()) { wordCount = wordCounts.next(); String sql = "insert into wordcount(word,count) " + "values('" + wordCount._1 + "'," + wordCount._2 + ")"; Statement stmt = conn.createStatement(); stmt.executeUpdate(sql); } // 用完以后，將連接還回去 ConnectionPool.returnConnection(conn); } }); return null; } }); jssc.start(); jssc.awaitTermination(); jssc.close(); } } ##運行腳本 [root@spark1 streaming]# cat persistWordCount.sh /usr/local/spark-1.5.1-bin-hadoop2.4/bin/spark-submit \ --class cn.spark.study.streaming.PersistWordCount \ --num-executors 3 \ --driver-memory 100m \ --executor-memory 100m \ --executor-cores 3 \ --files /usr/local/hive/conf/hive-site.xml \ --driver-class-path /usr/local/hive/lib/mysql-connector-java-5.1.17.jar \ /usr/local/spark-study/java/streaming/saprk-study-java-0.0.1-SNAPSHOT-jar-with-dependencies.jar \ ##運行nc [root@spark1 ~]# nc -lk 9999 hello word hello word hello java ##結果 mysql> use testdb; mysql> select * from wordcount; +----+---------------------+-------+-------+ | id | updated_time | word | count | +----+---------------------+-------+-------+ | 1 | 2019-08-19 14:52:45 | hello | 1 | | 2 | 2019-08-19 14:52:45 | word | 1 | | 3 | 2019-08-19 14:52:50 | hello | 2 | | 4 | 2019-08-19 14:52:50 | word | 2 | | 5 | 2019-08-19 14:52:55 | hello | 2 | | 6 | 2019-08-19 14:52:55 | word | 2 | | 7 | 2019-08-19 14:53:00 | hello | 2 | | 8 | 2019-08-19 14:53:00 | word | 2 | | 9 | 2019-08-19 14:53:05 | hello | 2 | | 10 | 2019-08-19 14:53:05 | word | 2 | | 11 | 2019-08-19 14:53:10 | hello | 2 | | 12 | 2019-08-19 14:53:10 | word | 2 | | 13 | 2019-08-19 14:53:15 | hello | 3 | | 14 | 2019-08-19 14:53:15 | word | 2 | | 15 | 2019-08-19 14:53:15 | java | 1 | | 16 | 2019-08-19 14:53:20 | hello | 3 | | 17 | 2019-08-19 14:53:20 | word | 2 | | 18 | 2019-08-19 14:53:20 | java | 1 | | 19 | 2019-08-19 14:53:25 | hello | 3 | | 20 | 2019-08-19 14:53:25 | word | 2 | | 21 | 2019-08-19 14:53:25 | java | 1 | +----+---------------------+-------+-------+ 21 rows in set (0.00 sec)