hdfs--Structured Streaming--console案例

先写一个python脚本用于在hdfs上生成文件

vim files.py

import osfor index in range(10):content = """{"name":"Leijun"}{"name":"Andy", "age":30}{"name":"Justin", "age":19}"""file_name = "/root/text{0}.json".format(index)with open(file_name, "w") as file:file.write(content)os.system("/usr/local/hadoop/bin/hdfs dfs -mkdir -p /structture")os.system("/usr/local/hadoop/bin/hdfs dfs -put {0} /structture".format(file_name))

Structured Streaming代码如下

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}object hdfsSource extends App {//构建spark sessionprivate val session: SparkSession = SparkSession.builder().master("local[2]").appName("test").getOrCreate()//创建schema,必须的一步private val schema = new StructType(Array(StructField("name", DataTypes.StringType),StructField("age", DataTypes.IntegerType)))//读取hdfs上的json文件,必须要有schema,不然报错private val df: DataFrame = session.readStream.schema(schema).json("hdfs://mypc01/structture")//输出读取的内容到控制台df.writeStream.outputMode(OutputMode.Update()).format("console").start().awaitTermination()
}

启动spark程序,同时执行python脚本

python files.py

输出结果

Batch: 0
-------------------------------------------+-------+----+
|   name| age|
+-------+----+
|Michael|null|
|   Andy|  30|
| Justin|  19|
|Michael|null|
|   Andy|  30|
| Justin|  19|
|Michael|null|
|   Andy|  30|
| Justin|  19|
...

解析

readStream

def readStream: DataStreamReader

返回一个DataStreamReader

   sparkSession.readStream.parquet("/path/to/directory/of/parquet/files")sparkSession.readStream.schema(schema).json("/path/to/directory/of/json/files")

schema

 def schema(schema: StructType): DataStreamReader

指定输入模式。某些数据源（例如JSON）可以根据数据自动推断输入架构。通过在此处指定架构，基础数据源可以跳过架构推断步骤，从而加快数据加载速度。
format

def format(source: String): DataStreamWriter[T]

指定基础输出数据源。

StructType

class StructType(fields: Array[StructField])
extends DataType
with Seq[StructField]

可以通过以下方式构造StructType对象：

 StructType(fields: Seq[StructField])

对于StructType对象，可以按名称提取一个或多个StructField。如果提取了多个StructField，则将返回一个StructType对象。如果提供的名称没有匹配的字段，它将被忽略。对于提取单个StructField的情况，将返回null。

总结

hdfs--Structured Streaming--console案例相关推荐

【Spark分布式内存计算框架——Structured Streaming】3. Structured Streaming —— 入门案例：WordCount
1.3 入门案例:WordCount 入门案例与SparkStreaming的入门案例基本一致:实时从TCP Socket读取数据(采用nc)实时进行词频统计WordCount,并将结果输出到控制台C ...
Structured Streaming 入门案例之WordCount
1.编写一个流式计算的应用, 不断的接收外部系统的消息 2.对消息中的单词进行词频统计 3.统计全局的结果步骤 Socket Server 等待 Structured Streaming 程序连接 ...
Structured Streaming基础入门
Structured Streaming 1. 回顾和展望 1.1. Spark 编程模型的进化过程 RDD rdd.flatMap(_.split(" ")).map((_, 1 ...
Structured Streaming 开发入门
Structured Streaming 作为 Spark 家族的新成员,通过 Spark SQL/DataFrame 来处理 Batch/Streaming 数据,基本的 SparkSQL API ...
2021年大数据Spark（五十）：Structured Streaming 案例一实时数据ETL架构
目录案例一实时数据ETL架构准备主题模拟基站日志数据实时增量ETL 案例一实时数据ETL架构在实际实时流式项目中,无论使用Storm.SparkStreami ...
Structured Streaming 案例初体验
Structured Streaming程序基本步骤编写Structured Streaming程序的基本步骤是: 1.创建SparkSession实例: 2.创建DataFrame表示从数据源输入 ...
Structured Streaming系列-1、Structured Streaming
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明. 传送门:大数据系列文章目录官方网址:http://spark.apache.org/. ht ...
大数据Spark Structured Streaming
目录 1 Spark Streaming 不足 2 Structured Streaming 概述 2.1 模块介绍 2.3 编程模型 3 入门案例:WordCount 3.1 功能演示 3.2 So ...
Structured Streaming编程 Programming Guide
Structured Streaming编程 Programming Guide • Overview • Quick Example • Programming Model o Basic Conc ...
2021年大数据Spark（五十二）：Structured Streaming 事件时间窗口分析
目录事件时间窗口分析时间概念 event-time 延迟数据处理延迟数据 Watermarking 水位官方案例演示事件 ...

hdfs--Structured Streaming--console案例

解析

总结

hdfs--Structured Streaming--console案例相关推荐

最新文章

热门文章