Structured Streaming教程(2) —— 常用输入与输出

上篇了解了一些基本的Structured Streaming的概念,知道了Structured Streaming其实是一个无下界的无限递增的DataFrame。基于这个DataFrame,我们可以做一些基本的select、map、filter操作,也可以做一些复杂的join和统计。本篇就着重介绍下,Structured Streaming支持的输入输出,看看都提供了哪些方便的操作。

数据源

Structured Streaming 提供了几种数据源的类型,可以方便的构造Steaming的DataFrame。默认提供下面几种类型:

File:文件数据源

file数据源提供了很多种内置的格式,如csv、parquet、orc、json等等,就以csv为例:

package xingoo.sstreaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType

object FileInputStructuredStreamingTest {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local")
      .appName("StructuredNetworkWordCount")
      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._
    val userSchema = new StructType().add("name", "string").add("age", "integer")
    val lines = spark.readStream
      .option("sep", ";")
      .schema(userSchema)
      .csv("file:///Users/xingoo/IdeaProjects/spark-in-action/data/*")

    val query = lines.writeStream
      .outputMode("append")
      .format("console")
      .start()

    query.awaitTermination()
  }
}

这样,在对应的目录下新建文件时,就可以在控制台看到对应的数据了。

aaa;1
bbb;2
aaa;5
ddd;6

还有一些其他可以控制的参数:

  • maxFilesPerTrigger 每个batch最多的文件数,默认是没有限制。比如我设置了这个值为1,那么同时增加了5个文件,这5个文件会每个文件作为一波数据,更新streaming dataframe。
  • latestFirst 是否优先处理最新的文件,默认是false。如果设置为true,那么最近被更新的会优先处理。这种场景一般是在监听日志文件的时候使用。
  • fileNameOnly 是否只监听固定名称的文件。

socket网络数据源

在我们自己练习的时候,一般都是基于这个socket来做测试。首先开启一个socket服务器,nc -lk 9999,然后streaming这边连接进行处理。

  spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

kafka数据源

这个是生产环境或者项目应用最多的数据源,通常架构都是:

应用数据输入-->kafka-->spark streaming -->其他的数据库

由于kafka涉及的内容还比较多,因此下一篇专门介绍kafka的集成。

输出

在配置完输入,并针对DataFrame或者DataSet做了一些操作后,想要把结果保存起来。就可以使用DataSet.writeStream()方法,配置输出需要配置下面的内容:

  • format : 配置输出的格式
  • output mode:输出的格式
  • query name:查询的名称,类似tempview的名字
  • trigger interval:触发的间隔时间,如果前一个batch处理超时了,那么不会立即执行下一个batch,而是等下一个trigger时间在执行。
  • checkpoint location:为保证数据的可靠性,可以设置检查点保存输出的结果。

output Mode

详细的来看看这个输出模式的配置,它与普通的Spark的输出不同,只有三种类型:

  • complete,把所有的DataFrame的内容输出,这种模式只能在做agg聚合操作的时候使用,比如ds.group.count,之后可以使用它
  • append,普通的dataframe在做完map或者filter之后可以使用。这种模式会把新的batch的数据输出出来,
  • update,把此次新增的数据输出,并更新整个dataframe。有点类似之前的streaming的state处理。

输出的类型

Structed Streaming提供了几种输出的类型:

  • file,保存成csv或者parquet
noAggDF
  .writeStream
  .format("parquet")
  .option("checkpointLocation", "path/to/checkpoint/dir")
  .option("path", "path/to/destination/dir")
  .start()
  • console,直接输出到控制台。一般做测试的时候用这个比较方便。
noAggDF
  .writeStream
  .format("console")
  .start()
  • memory,可以保存在内容,供后面的代码使用
aggDF
  .writeStream
  .queryName("aggregates")
  .outputMode("complete")
  .format("memory")
  .start()
spark.sql("select * from aggregates").show()  
  • foreach,参数是一个foreach的方法,用户可以实现这个方法实现一些自定义的功能。
writeStream
    .foreach(...)
    .start()

这个foreach的功能很强大,稍后也会详细的说明。

原文地址:https://www.cnblogs.com/xing901022/p/9135257.html

时间: 2024-10-10 10:55:13

Structured Streaming教程(2) —— 常用输入与输出的相关文章

Structured Streaming教程(1) —— 基本概念与使用

近年来,大数据的计算引擎越来越受到关注,spark作为最受欢迎的大数据计算框架,也在不断的学习和完善中.在Spark2.x中,新开放了一个基于DataFrame的无下限的流式处理组件--Structured Streaming,它也是本系列的主角,废话不多说,进入正题吧! 简单介绍 在有过1.6的streaming和2.x的streaming开发体验之后,再来使用Structured Streaming会有一种完全不同的体验,尤其是在代码设计上. 在过去使用streaming时,我们很容易的理解

Structured Streaming教程(3) —— 与Kafka的集成

Structured Streaming最主要的生产环境应用场景就是配合kafka做实时处理,不过在Strucured Streaming中kafka的版本要求相对搞一些,只支持0.10及以上的版本.就在前一个月,我们才从0.9升级到0.10,终于可以尝试structured streaming的很多用法,很开心~ 引入 如果是maven工程,直接添加对应的kafka的jar包即可: <dependency> <groupId>org.apache.spark</groupI

9.常用输入与输出与确认

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> <html xmlns="http://www.w3.org/1999/xhtml"> <head> <meta http-equiv="Content-

Spark Structured Streaming框架(3)之数据输出源详解

Spark Structured streaming API支持的输出源有:Console.Memory.File和Foreach.其中Console在前两篇博文中已有详述,而Memory使用非常简单.本文着重介绍File和Foreach两种方式,并介绍如何在源码基本扩展新的输出方式. 1. File Structured Streaming支持将数据以File形式保存起来,其中支持的文件格式有四种:json.text.csv和parquet.其使用方式也非常简单只需设置checkpointLo

Spark Structured Streaming框架(2)之数据输入源详解

Spark Structured Streaming目前的2.1.0版本只支持输入源:File.kafka和socket. 1. Socket Socket方式是最简单的数据输入源,如Quick example所示的程序,就是使用的这种方式.用户只需要指定"socket"形式并配置监听的IP和Port即可. val scoketDF = spark.readStream .format("socket") .option("host","

Shell编程基础教程3--Shell输入与输出

3.Shell输入与输出    3.1.echo        echo命令可以显示文本行或变量,或者把字符串输出到文件        echo [option] string            -e:解析转义字符            -n:回车不换行,linux系统默认回车换行            转义字符:\c,\f,\t,\n……        例子程序 #!/bin/bash #echo echo -e "This echo's 3 new lines\n\n\n" e

Apache Spark 2.2.0 中文文档 - Structured Streaming 编程指南 | ApacheCN

Structured Streaming 编程指南 概述 快速示例 Programming Model (编程模型) 基本概念 处理 Event-time 和延迟数据 容错语义 API 使用 Datasets 和 DataFrames 创建 streaming DataFrames 和 streaming Datasets Input Sources (输入源) streaming DataFrames/Datasets 的模式接口和分区 streaming DataFrames/Dataset

学习Spark2.0中的Structured Streaming(一)

转载自:http://lxw1234.com/archives/2016/10/772.htm Spark2.0新增了Structured Streaming,它是基于SparkSQL构建的可扩展和容错的流式数据处理引擎,使得实时流式数据计算可以和离线计算采用相同的处理方式(DataFrame&SQL).Structured Streaming顾名思义,它将数据源和计算结果都映射成一张"结构化"的表,在计算的时候以结构化的方式去操作数据流,大大方便和提高了数据开发的效率. Sp

【Python基础】python控制台输入、输出

python控制台输入.输出 1.输出 简单输出 print("我是简单的字符串输出") 控制台运行结果 我是简单的字符串输出 格式化输出 age=18 print("我今年%d岁"%age) age+=1 print("我今年%d岁"%age) age+=1 print("我今年%d岁"%age) 控制台打印结果: 我今年18岁 我今年19岁 我今年20岁 常用格式化符号: 格式符号 转换 %c 字符 %s 等同于str()