Spark结构化数据流StructuredStreaming.docxVIP

Spark结构化数据流StructuredStreaming是一个构建在Spark SQL之上的一个高容错可扩展的流处理引擎。当然你的流计算也可以用相同的方式对静态数据进行批处理计算。它可以不断更新持续流进来的递增数据，并且将计算结果也持续的更新。目前的Dataset/DataFrame的API支持的语言有Scala,Java和Python三种语言。可以实现流聚合，事件窗口，Join聚合等。结构化流处理是建立在Spark SQL优化引擎Catylist之上的又一引擎。所以，其性能也是非常好的。未来的发展的空间也是巨大的。该种流处理保证有且只处理数据一次，同时实现端到端的数据。通过检查点checkpoint和Write Ahead Logs机制实现高容错。在Spark2.1版本中，其API仍然还是试验性质的,接下来我们就来一起走进StructuredStreaming的殿堂吧。import org.apache.spark.sql.SparkSessionimport org.apache.spark.SparkConfobjectTest {defmain(args: Array[String]): Unit = {/* * 配置信息：配置应用名称*StructuredStreaming Test * Master为local[3] */valconf = new SparkConf().setAppName(StructuredStreaming Test).setMaster(local[3])valspark =SparkSession.builder().config(conf).getOrCreate()import spark.implicits._//获取一行一行的数据，注意，此时的host对应的、//master为数据源那台机器的主机名//并且已经配置好了hosts文件里的ip与主机映射//此时的lines是一个DataFrame对象vallines = spark .readStream .format(socket) .option(host, master) .option(port, 9999) .load()//用Dataset里的flatMap方法将每行转换为一个一//个的单词valwords = lines.as[String].flatMap(_.split( ))//单词计数valwordCounts = words.groupBy(value).count()//返回一个流查询对象valquery = wordCounts.writeStream.outputMode(complet).format(console).start()//等待终端query.awaitTermination() }}图1-1 StructuredNetworkWordCount代码实现以上这段代码是处理的数据源是一台主机名为master的9999端口。通过监听TCP端口实时监听获取流数据源。因此，我们需要通过以下的方式来开启TCP端口的监听nc -lk 9999通过观察console的终端输出的结果，我们可以判定，这段程序处理的是以10秒为单位的微批数据，也就是说，每隔10秒钟进行一次计算。此外，你还可以通过Spark官方提供的案例来运行你的第一个结构化流处理的应用程序。在Spark官方提供的二进制包里面的bin目录下有一个run-examples的脚本，当加上org.apache.spark.examples.sql.streaming.StructuredNetworkWordCount的参数后，运行这个脚本，这样也能运行您的第一个结构化流处理程序。运行了第一个结构化流程序之后，我们再回过头来看看Spark源码中提供给我们的三类案例。这三类案例在Spark源码包中的路径为examples/src/main/java/org.apache.spark.sql.streaming/streaming和examples/src/main/scala/org.apache.spark.sql.streaming/streaming（源码的导入方法详见第xxx章第xxx节）在Java代码包路径下，一共有三个案例，分别为JavaStructuredNetworkWordCount,JavaStructuredKafkaworkWordCount,JavaStructuredNetworkWordCountWindowed。在Scala代码包路径下，也是有三个案例，分别为StructuredNetworkWordCount,StructuredKafkaworkWo