14.DStream的output操作以及foreachRDD详解

output操作概览

Output

Meaning

print

打印每个batch中的前10个元素，主要用于测试，或者是不需要执行什么output操作时，用于简单触发一下job。

saveAsTextFile(prefix, [suffix])

将每个batch的数据保存到文件中。每个batch的文件的命名格式为：prefix-TIME_IN_MS[.suffix]

saveAsObjectFile

同上，但是将每个batch的数据以序列化对象的方式，保存到SequenceFile中。

saveAsHadoopFile

同上，将数据保存到Hadoop文件中

foreachRDD

最常用的output操作，遍历DStream中的每个产生的RDD，进行处理。可以将每个RDD中的数据写入外部存储，比如文件、数据库、缓存等。通常在其中，是针对RDD执行action操作的，比如foreach。

output操作

 

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

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

foreachRDD详解

通常在foreachRDD中，都会创建一个Connection，比如JDBC Connection，然后通过Connection将数据写入外部存储。

误区一：在RDD的foreach操作外部，创建Connection

这种方式是错误的，因为它会导致Connection对象被序列化后传输到每个Task中。而这种Connection对象，实际上一般是不支持序列化的，也就无法被传输。

  1. dstream.foreachRDD {rdd =>
  2.   val connection = createNewConnection()
  3.   rdd.foreach { record =>connection.send(record)
  4.   }
  5. }

误区二：在RDD的foreach操作内部，创建Connection

这种方式是可以的，但是效率低下。因为它会导致对于RDD中的每一条数据，都创建一个Connection对象。而通常来说，Connection的创建，是很消耗性能的。

  1. dstream.foreachRDD {rdd =>
  2.   rdd.foreach { record =>
  3.     val connection = createNewConnection()
  4.     connection.send(record)
  5.     connection.close()
  6.   }
  7. }

合理方式一：使用RDD的foreachPartition操作，并且在该操作内部，创建Connection对象，这样就相当于是，为RDD的每个partition创建一个Connection对象，节省资源的多了。

  1. dstream.foreachRDD {rdd =>
  2.   rdd.foreachPartition { partitionOfRecords=>
  3.     val connection = createNewConnection()
  4.     partitionOfRecords.foreach(record =>connection.send(record))
  5.     connection.close()
  6.   }
  7. }

合理方式二：自己手动封装一个静态连接池，使用RDD的foreachPartition操作，并且在该操作内部，从静态连接池中，通过静态方法，获取到一个连接，使用之后再还回去。这样的话，甚至在多个RDD的partition之间，也可以复用连接了。而且可以让连接池采取懒创建的策略，并且空闲一段时间后，将其释放掉。

  1. dstream.foreachRDD {rdd =>
  2.   rdd.foreachPartition { partitionOfRecords=>
  3.     val connection = ConnectionPool.getConnection()
  4.     partitionOfRecords.foreach(record =>connection.send(record))
  5.     ConnectionPool.returnConnection(connection) 
  6.   }
  7. }

foreachRDD实战

 

案例：改写UpdateStateByKeyWordCount，将每次统计出来的全局的单词计数，写入一份，到MySQL数据库中。

建表语句

  1. create table wordcount(
  2.   id integer auto_increment primary key,
  3.   updated_time timestamp NOT NULL defaultCURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
  4.   word varchar(255),
  5.   count integer
  6. );

java版本代码：

  1. package cn.spark.study.sql;
  2.  
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.Statement;
  6. import java.util.ArrayList;
  7. import java.util.HashMap;
  8. import java.util.List;
  9. import java.util.Map;
  10.  
  11. import org.apache.spark.SparkConf;
  12. import org.apache.spark.api.java.JavaPairRDD;
  13. import org.apache.spark.api.java.JavaRDD;
  14. import org.apache.spark.api.java.JavaSparkContext;
  15. import org.apache.spark.api.java.function.Function;
  16. import org.apache.spark.api.java.function.PairFunction;
  17. import org.apache.spark.api.java.function.VoidFunction;
  18. import org.apache.spark.sql.DataFrame;
  19. import org.apache.spark.sql.Row;
  20. import org.apache.spark.sql.RowFactory;
  21. import org.apache.spark.sql.SQLContext;
  22. import org.apache.spark.sql.types.DataTypes;
  23. import org.apache.spark.sql.types.StructField;
  24. import org.apache.spark.sql.types.StructType;
  25. import scala.Tuple2;
  26.  
  27. /**
  28.  * JDBC数据源
  29.  * @author leizq120310
  30.  *
  31.  */
  32.  
  33. public class JDBCDataSource {
  34.  
  35. 	public static void main(String[] args) {
  36. 		SparkConf conf = new SparkConf()
  37. 				.setAppName("JDBCDataSource");
  38. 		JavaSparkContext sc = new JavaSparkContext(conf);
  39. 		SQLContext sqlContext = new SQLContext(sc);
  40. 		
  41. 		// 总结一下
  42. 		// jdbc数据源
  43. 		// 首先，是通过SQLContext的read系列方法，将mysql中的数据加载为DataFrame
  44. 		// 然后可以将DataFrame转换为RDD，使用Spark Core提供的各种算子进行操作
  45. 		// 最后可以将得到的数据结果，通过foreach()算子，写入mysql、hbase、redis等等db/ cache中
  46. 		
  47. 		// 分别将mysql中两张表的数据加载为DataFrame
  48. 		Map options = new HashMap();
  49. 		options.put("url", "jdbc:mysql://spark1:3306/testdb");
  50. 		options.put("dbtable", "student_infos");
  51. 		DataFrame studentInfosDF = sqlContext.read().format("jdbc").options(options).load();
  52. 		
  53. 		options.clear();
  54. 		options.put("url", "jdbc:mysql://spark1:3306/testdb");
  55. 		options.put("dbtable", "student_scores");
  56. 		DataFrame studentScoresDF = sqlContext.read().format("jdbc").options(options).load();
  57. 		
  58. 		// 将两个DataFrame转换为JavaPairRDD，执行join操作
  59. 		JavaPairRDD> studentsRDD = studentInfosDF.javaRDD().mapToPair(new PairFunction() {
  60. 				private static final long serialVersionUID = 1L;
  61. 	
  62. 				@Override
  63. 				public Tuple2 call(Row row) throws Exception {
  64. 					return new Tuple2
  65. 					(row.getString(0), Integer.valueOf(String.valueOf(row.get(1))));
  66. 				}
  67. 			}).join(studentScoresDF.javaRDD().mapToPair(new PairFunction() {
  68. 	
  69. 				private static final long serialVersionUID = 1L;
  70. 	
  71. 				@Override
  72. 				public Tuple2 call(Row row) throws Exception {
  73. 					return new Tuple2
  74. 					(row.getString(0), Integer.valueOf(String.valueOf(row.get(1))));
  75. 				}
  76. 			}));
  77. 		
  78. 		// 将JavaPairRDD转换为JavaRDD
  79. 		JavaRDD studentRowsRDD = studentsRDD.map(new Function>, Row>() {
  80.  
  81. 			private static final long serialVersionUID = 1L;
  82.  
  83. 			@Override
  84. 			public Row call(Tuple2> tuple) throws Exception {
  85. 				return RowFactory.create(tuple._1, tuple._2._1, tuple._2._2);
  86. 			}
  87. 		});
  88. 		
  89. 		// 过滤出分数大于80分的数据
  90. 		JavaRDD filteredStudentRowsRDD = studentRowsRDD.filter(new Function() {
  91.  
  92. 			private static final long serialVersionUID = 1L;
  93.  
  94. 			@Override
  95. 			public Boolean call(Row row) throws Exception {
  96. 				if (row.getInt(2) > 80) {
  97. 					return true;
  98. 				}
  99. 				return false;
  100. 			}
  101. 		});
  102. 		
  103. 		// 转换为DataFrame
  104. 		List structFields = new ArrayList();
  105. 		structFields.add(DataTypes.createStructField("name", DataTypes.StringType, true));
  106. 		structFields.add(DataTypes.createStructField("age", DataTypes.IntegerType, true));
  107. 		structFields.add(DataTypes.createStructField("score", DataTypes.IntegerType, true));
  108. 		StructType structType = DataTypes.createStructType(structFields);
  109. 		
  110. 		DataFrame studentsDF = sqlContext.createDataFrame(filteredStudentRowsRDD, structType);
  111. 		
  112. 		Row[] rows = studentsDF.collect();
  113. 		for (Row row : rows)
  114. 		{
  115. 			System.out.println(row);
  116. 		}
  117. 		
  118. 		// 将DataFrame中的数据保存到mysql表中
  119. 		// 这种方式是在企业里很常用的，有可能是插入mysql、有可能是插入hbase，还有可能是插入redis缓存
  120. 		studentsDF.javaRDD().foreach(new VoidFunction() {
  121. 			private static final long serialVersionUID = 1L;
  122.  
  123. 			@Override
  124. 			public void call(Row row) throws Exception {
  125. 				String sql = "insert into good_student_infos values("
  126. 						+ "'" + String.valueOf(row.getString(0)) + "',"
  127. 						+ Integer.valueOf(String.valueOf(row.get(1))) + ","
  128. 						+ Integer.valueOf(String.valueOf(row.get(2))) + ")";
  129. 				Class.forName("com.mysql.jdbc.Driver");
  130. 				
  131. 				Connection conn = null;
  132. 				Statement stmt = null;
  133. 				try{
  134. 					conn = DriverManager.getConnection(
  135. 							"jdbc:mysql://spark1:3306/testdb", "", "");
  136. 					stmt = conn.createStatement();
  137. 					stmt.executeUpdate(sql);
  138. 				} catch (Exception e) {
  139. 					e.printStackTrace();
  140. 				} finally{
  141. 					if (stmt != null){
  142. 						stmt.close();
  143. 					}
  144. 					if (conn != null){
  145. 						conn.close();
  146. 					}
  147. 				}
  148. 			}
  149. 		});
  150. 		sc.close(); 
  151. 	}
  152. }

操作步骤：

 

1.进入linux系统平台下的mysql操作中，创建wordcount表

文章最后，给大家推荐一些受欢迎的技术博客链接：

  1. Hadoop相关技术博客链接
  2. Spark 核心技术链接
  3. 超全干货--Flink思维导图，花了3周左右编写、校对
  4. 深入JAVA 的JVM核心原理解决线上各种故障【附案例】
  5. 请谈谈你对volatile的理解？--最近小李子与面试官的一场“硬核较量”
  6. 聊聊RPC通信，经常被问到的一道面试题。源码+笔记，包懂

 

欢迎扫描下方的二维码或 搜索 公众号“10点进修”，我们会有更多、且及时的资料推送给您，欢迎多多交流！

                                           

       

 

 

 

注：本文转载自blog.csdn.net的不埋雷的探长的文章"https://blog.csdn.net/weixin_32265569/article/details/78578776"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！