AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Structured Streaming中DS和DF的操作详解

2022-05-15 197
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 笔记

创建流式的DataSet和DataFrame

方式一：通过JavaBean方式

Java语言实现：

package com.kfk.spark.structuredstreaming;
import java.sql.Date;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/21
 * @time : 10:38 下午
 */
public class DeviceData {
    // device: string, type: string, signal: double, time: DateType
    private String device;
    private String deviceType;
    private double signal;
    private Date deviceTime;
    public String getDevice() {
        return device;
    }
    public void setDevice(String device) {
        this.device = device;
    }
    public String getDeviceType() {
        return deviceType;
    }
    public void setDeviceType(String deviceType) {
        this.deviceType = deviceType;
    }
    public double getSignal() {
        return signal;
    }
    public void setSignal(double signal) {
        this.signal = signal;
    }
    public Date getDeviceTime() {
        return deviceTime;
    }
    public void setDeviceTime(Date deviceTime) {
        this.deviceTime = deviceTime;
    }
    public DeviceData(String device, String deviceType, double signal, Date deviceTime) {
        this.device = device;
        this.deviceType = deviceType;
        this.signal = signal;
        this.deviceTime = deviceTime;
    }
    public DeviceData(){
    }
}

package com.kfk.spark.structuredstreaming;
import com.kfk.spark.common.CommSparkSession;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import java.sql.Date;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/21
 * @time : 10:35 下午
 */
public class StruStreamingDFOper {
    public static void main(String[] args) throws StreamingQueryException {
        SparkSession spark = CommSparkSession.getSparkSession();
        // input table
        Dataset<String> dflines = spark.readStream()
                .format("socket")
                .option("host", "bigdata-pro-m04")
                .option("port", 9999)
                .load().as(Encoders.STRING());
        // 根据javaBean转换为dataset
        Dataset<DeviceData> dfdevice = dflines.map(new MapFunction<String, DeviceData>() {
            @Override
            public DeviceData call(String value) throws Exception {
                String[] lines = value.split(",");
                return new DeviceData(lines[0],lines[1],Double.parseDouble(lines[2]), Date.valueOf(lines[3]));
            }
        }, ExpressionEncoder.javaBean(DeviceData.class));
        // result table
        Dataset<Row> dffinal = dfdevice.select("device","deviceType").where("signal > 10").groupBy("deviceType").count();
        // output
        StreamingQuery query = dffinal.writeStream()
                .outputMode("update")
                .format("console")
                .start();
        query.awaitTermination();
    }
}

方式二：构造Schema的方式

package com.kfk.spark.structuredstreaming;
import com.kfk.spark.common.CommSparkSession;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;
import org.apache.spark.sql.catalyst.encoders.RowEncoder;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.sql.Date;
import java.util.ArrayList;
import java.util.List;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/21
 * @time : 10:35 下午
 */
public class StruStreamingDFOper2 {
    public static void main(String[] args) throws StreamingQueryException {
        SparkSession spark = CommSparkSession.getSparkSession();
        // input table
        Dataset<String> dflines = spark.readStream()
                .format("socket")
                .option("host", "bigdata-pro-m04")
                .option("port", 9999)
                .load().as(Encoders.STRING());
        List<StructField> fields = new ArrayList<StructField>();
        StructField device = DataTypes.createStructField("device",DataTypes.StringType,true);
        StructField deviceType = DataTypes.createStructField("deviceType",DataTypes.StringType,true);
        StructField signal = DataTypes.createStructField("signal",DataTypes.DoubleType,true);
        StructField deviceTime = DataTypes.createStructField("deviceTime",DataTypes.DateType,true);
        fields.add(device);
        fields.add(deviceType);
        fields.add(signal);
        fields.add(deviceTime);
        // 构造Schema
        StructType scheme = DataTypes.createStructType(fields);
        // 根据schema转换为dataset
        Dataset<Row> dfdevice = dflines.map(new MapFunction<String, Row>() {
            @Override
            public Row call(String value) throws Exception {
                String[] lines = value.split(",");
                return RowFactory.create(lines[0],lines[1],Double.parseDouble(lines[2]), Date.valueOf(lines[3]));
            }
        }, RowEncoder.apply(scheme));
        // result table
        Dataset<Row> dffinal = dfdevice.select("device","deviceType").where("signal > 10").groupBy("deviceType").count();
        // output
        StreamingQuery query = dffinal.writeStream()
                .outputMode("update")
                .format("console")
                .start();
        query.awaitTermination();
    }
}

Scala语言实现：

package com.kfk.spark.structuredstreaming
import com.kfk.spark.common.CommSparkSessionScala
import java.sql.Date
import org.apache.spark.sql.{Dataset, Row}
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.types.StructType;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/22
 * @time : 7:31 下午
 */
object StruStreamingDFOperScala {
    case  class  DeviceData(device : String,deviceType : String , signal : Double ,deviceTime: Date)
    def main(args: Array[String]): Unit = {
        val spark = CommSparkSessionScala.getSparkSession() ;
        import spark.implicits._;
        // input table
        val socketDF = spark
                .readStream
                .format("socket")
                .option("host", "bigdata-pro-m04")
                .option("port", 9999)
                .load().as[String]
        // 构造Schema
        val userSchema = new StructType()
                .add("device", "string")
                .add("deviceType", "string")
                .add("signal", "double")
                .add("deviceTime", "date")
        // 根据schema转换为dataset
        val df_device = socketDF.map(x => {
            val lines = x.split(",")
            Row(lines(0),lines(1),lines(2),lines(3))
        })(RowEncoder(userSchema))
        // 根据case  class转换为dataset
        val df : Dataset[DeviceData] = df_device.as[DeviceData]
        // result table
        val df_final = df.groupBy("deviceType").count()
        // output
        val query = df_final.writeStream
                .outputMode("update")
                .format("console")
                .start()
        query.awaitTermination()
    }
}

测试数据：

aa,ss,100,2020-12-21
bb,ss,47,2020-12-21
cc,ss,3,2020-12-21
dd,ss,46,2020-12-21

运行结果：

-------------------------------------------
Batch: 1
-------------------------------------------
+----------+-----+
|deviceType|count|
+----------+-----+
|        ss|    3|
+----------+-----+

创建流式的DataSet和DataFrame也可以通过创建临时表来进行业务分析

package com.kfk.spark.structuredstreaming;
import com.kfk.spark.common.CommSparkSession;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;
import org.apache.spark.sql.catalyst.encoders.RowEncoder;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import java.sql.Date;
import java.util.ArrayList;
import java.util.List;
/**
 * @author : 蔡政洁
 * @email :caizhengjie888@icloud.com
 * @date : 2020/12/21
 * @time : 10:35 下午
 */
public class StruStreamingDFOper3 {
    public static void main(String[] args) throws StreamingQueryException {
        SparkSession spark = CommSparkSession.getSparkSession();
        // input table
        Dataset<String> dflines = spark.readStream()
                .format("socket")
                .option("host", "bigdata-pro-m04")
                .option("port", 9999)
                .load().as(Encoders.STRING());
        List<StructField> fields = new ArrayList<StructField>();
        StructField device = DataTypes.createStructField("device",DataTypes.StringType,true);
        StructField deviceType = DataTypes.createStructField("deviceType",DataTypes.StringType,true);
        StructField signal = DataTypes.createStructField("signal",DataTypes.DoubleType,true);
        StructField deviceTime = DataTypes.createStructField("deviceTime",DataTypes.DateType,true);
        fields.add(device);
        fields.add(deviceType);
        fields.add(signal);
        fields.add(deviceTime);
        // 构造Schema
        StructType scheme = DataTypes.createStructType(fields);
        // 根据schema转换为dataset
        Dataset<Row> dfdevice = dflines.map(new MapFunction<String, Row>() {
            @Override
            public Row call(String value) throws Exception {
                String[] lines = value.split(",");
                return RowFactory.create(lines[0],lines[1],Double.parseDouble(lines[2]), Date.valueOf(lines[3]));
            }
        }, RowEncoder.apply(scheme));
        // 创建临时表
        dfdevice.createOrReplaceTempView("device");
        // result table
        Dataset<Row> dffinal = spark.sql("select * from device");
        // output
        StreamingQuery query = dffinal.writeStream()
                .outputMode("append")
                .format("console")
                .start();
        query.awaitTermination();
    }
}
文章标签：
Java
流计算
Scala
游客wkxim4agoo6le
目录
相关文章
九芒星#
|
运维 监控 安全
【网络安全】护网系列-社工&溯源
【网络安全】护网系列-社工&溯源
九芒星#
1410 0
Echo_Wish
|
8月前
|
传感器 人工智能 物联网
穿戴科技新风尚：智能服装设计与技术全解析
穿戴科技新风尚：智能服装设计与技术全解析
Echo_Wish
693 85
Lux_Sun
|
Web App开发 移动开发 JavaScript
JS - 微信浏览器（H5）语音录音插件（Recorder H5）
JS - 微信浏览器（H5）语音录音插件（Recorder H5）
Lux_Sun
3070 0
aliyun9879145816
【ChatGLM】本地版ChatGPT ？6G显存即可轻松使用 ！ChatGLM-6B 清华开源模型本地部署教程
【ChatGLM】本地版ChatGPT ？6G显存即可轻松使用 ！ChatGLM-6B 清华开源模型本地部署教程
aliyun9879145816
932 0
Yule
|
8月前
|
网络协议 安全
修复安全组未开放风险端口
安全体检结果显示，安全组22/3389端口TCP/UDP协议的入方向规则目的地址设置为0.0.0.0/0存在风险，易遭暴力破解。修复过程：修改安全组22的授权对象，限制访问源地址，增强服务器登录安全性，有效降低风险。
Yule
254 13
程序员Feri
|
9月前
|
人工智能 程序员 iOS开发
一文彻底学会HarmonyOS的AI编程助手
本文介绍了华为官方AI辅助编程工具CodeGenie，该工具支持HarmonyOS NEXT领域的智能问答、ArkTS代码补全/生成及万能卡片生成，显著提升开发效率。安装步骤包括下载插件、离线安装及授权登录，功能涵盖知识问答、代码补全与生成、以及智能生成HarmonyOS万能卡片。
程序员Feri
355 0
1941623231718325
|
11月前
|
人工智能 搜索推荐 安全
数字孪生与教育：虚拟实验室的兴起
数字孪生技术通过模拟、分析和优化，为教育创新提供了新机遇。特别是在虚拟实验室的构建和应用上，数字孪生技术打破了物理限制，提供了丰富的学习体验，支持精准教学与个性化学习，有效培养学生的创新能力和实践能力。国内外高校已积极应用，未来将更加智能化、个性化。
1941623231718325
231 3
vohelon
|
存储 缓存 算法
什么是配置中心页面？
【10月更文挑战第24天】什么是配置中心页面？
vohelon
260 3
游客rvfuxjwu5pc2e
|
机器学习/深度学习 人工智能 JSON
在LLM浪潮下，prompt工程师需要很懂算法吗？
最近AI大神吴恩达推出prompt教程并给出了prompt构建三大原则，Prompt Engineering Guide也提出了诸多技巧，受AI技术的快速发展的影响，你觉得在LLM浪潮下，prompt工程师需要很懂算法吗？欢迎一起来聊一聊～
游客rvfuxjwu5pc2e
26058 50
在LLM浪潮下，prompt工程师需要很懂算法吗？
帐篷Li
|
IDE 开发工具 流计算
保姆级Arduino开发环境搭建
保姆级Arduino开发环境搭建
帐篷Li
402 1

热门文章

最新文章

  • 1
    linux下ffmpeg安装
  • 2
    overleaf 插入图片，引用图片，图标标题Fig与文章引用Figure不一致解决
  • 3
    364 页 PyTorch 版《动手学深度学习》PDF 开源了（全中文，支持 Jupyter 运行）
  • 4
    阿里云上的几种数据备份方式
  • 5
    用深度神经网络修复H漫：看完这篇你就能眼中无码
  • 6
    MySQL · 引擎特性 · 像NOSQL那样使用MySQL
  • 7
    解决GitHub文件无法下载的问题
  • 8
    OpenSearch大模型实践之Havenask篇
  • 9
    SpringBoot整合Java Mail实现Outlook / Office365发送邮件
  • 10
    2019阿里云双11来袭,价格非常重要,但是懂得上云攻略更重要
  • 1
    AI Agent框架
    62
  • 2
    AI Agent：从工具到伙伴的智能进化
    72
  • 3
    计算机网络--TCP和UDP学习
    30
  • 4
    LBA-ECO ND-30 营养分析和气体通量，森林年代序列，巴西帕拉州
    30
  • 5
    2025年测试工程师的核心能力：掌握Dify工作流编排AI测试智能体
    60
  • 6
    有了解过 SpringBoot 的参数配置吗？
    53
  • 7
    分类模型校准：ROC-AUC不够？用ECE/pMAD评估概率质量
    24
  • 8
    把服务器当代码写：IaC 让部署变得又稳又香
    49
  • 9
    SDN 和 NFV：把网络“拆开重做”的底层革命，到底牛在哪？
    54
  • 10
    用数据“掘地三尺”，为零碳排放找出一条最靠谱的路
    32

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    阿里云弹性公网IP线路类型【BGP(多线)_精品】是什么意思？