使用Spark SQL进行流式机器学习计算(上)

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 今天来和大家简单说一下如何使用Spark SQL进行流式数据的机器学习处理

今天来和大家聊一下如何使用Spark SQL进行流式数据的机器学习处理。本文主要分为以下几个章节:

  • 什么是流式机器学习
  • 机器学习模型获取途径
  • 系统演示

1. 什么是流式机器学习

通常,当我们听到有人提到实时数据机器学习时,其实他们是讨论:

  • 他们希望有一个模型,这个模型利用最近历史信息来进行预测分析。举一个天气的例子,如果最近几天都是晴天,那么未来几天极小概率会出现雨雪和低温天气
  • 这个模型还需要是可更新的。当数据流经系统时,模型是可以随之进化升级。举个例子,随着业务规模的扩大,我们希望零售销售模型仍然保持准确。

第一个例子我们可以将它归为时序预测。第二个例子中,模型需要更新或者重新训练,这是一个non-stationarity问题。时序预测和non-stationarity数据分布是两类不同的问题。本文主要关注第二类问题,对于这类问题,一般的解决方案主要有:

  • 增量式算法:有一些算法支持通过数据逐步学习。也就是说,每次进来一些新的数据时,模型会被更新。SVM,神经网络等算法都有增量式版本,此外贝叶斯网络也可以用作增量学习。
  • 周期重新学习:一个更加直接的方法就是用一批最新数据重新训练我们的模型。这种方法可以用到的绝大多数的算法上。

2. 机器学习模型获取途径

实时机器学习应用分成两块,一部分是模型实时训练,另一部分是数据实时预测分析。现实中,我们可能没法实现模型的实时训练,只能退而求其次地使用已经训练好模型。这些模型可能会周期性地使用历史数据训练更新一次。所以,我们可以根据实际的算法和模型时效性要求,来选择实时训练模型还是使用预训练好的模型。

  • 模型算法支持增量训练:可以选择用流式数据实时训练更新
  • 模型算法不自持增量训练:可以选择用离线数据预先训练好模式

回到主题上,我们要实现使用Spark SQL进行流式机器学习。前面几篇文章已经简单介绍了EMR如何使用Spark SQL进行流式ETL处理。既然要进行机器学习,我们很自然地想到Spark MLlib。如果想将Spark MLlib应用到Spark SQL上,我们可以简单地将MLlib算法包装成UDF使用。另外一个模型获取途径是利用阿里云上的一些在线机器学习服务,我们可以将在线机器学习服务使用UDF封装后使用。

  • 使用UDF封装现有的Spark MLlib算法
  • 使用UDF封装阿里云在线机器学习服务

限于篇幅,我会分两篇文章分别介绍这两个方式,本文将简单介绍如何利用Spark MLlib进行流式机器学习。

3. 系统演示

本节,我们将演示一下如何利用逻辑回归算法进行演示。

3.1 系统架构

下面这张图展示了整个实时监测系统的架构,前端接LogService数据,实时监测分析结果写入到RDS,最后通过DataV展示出来。
image

3.2 测试数据集

测试数据集使用Spark自带的sample_libsvm_data.txt,我们要做的是写一个数据生成器,将数据集的数据不断地向SLS中发送,模拟流式数据。

算法模型准备

Spark MLlib提供了大量的机器学习算法实现,可以方便的再RDD或者DataFrame API上使用,但是无法直接用在SQL API上,所以我们需要使用UDF来封装一下。这里,我们选用逻辑回归算法,具体的实现就不细说了,可以参考这里的代码:LogisticRegressionUDF.scala

3.4 部署测试

  • CLI
## emr datasources包还没有发布,需要手动编译出来
git clone git@github.com:aliyun/aliyun-emapreduce-sdk.git
cd aliyun-emapreduce-sdk
git checkout -b master-2.x origin/master-2.x
mvn clean package -DskipTests

## 编译完后, assembly/target目录下会生成emr-datasources_shaded_2.11-1.7.0-SNAPSHOT.jar

spark-sql --master yarn-client --num-executors 2 --executor-memory 2g --executor-cores 2 --jars emr-datasources_shaded_2.11-1.7.0-SNAPSHOT.jar --driver-class-path emr-datasources_shaded_2.11-1.7.0-SNAPSHOT.jar
  • 建表
spark-sql> CREATE DATABASE IF NOT EXISTS default;
spark-sql> USE default;

-- 测试数据源
spark-sql> CREATE TABLE IF NOT EXISTS sls_dataset
USING loghub
OPTIONS (
sls.project = "${logProjectName}",
sls.store = "${logStoreName}",
access.key.id = "${accessKeyId}",
access.key.secret = "${accessKeySecret}",
endpoint = "${endpoint}");

spark-sql> DESC sls_dataset
__logProject__  string  NULL
__logStore__  string  NULL
__shard__ int NULL
__time__  timestamp NULL
__topic__ string  NULL
__source__  string  NULL
label string  NULL
features  string  NULL
__tag__hostname__ string  NULL
__tag__path__ string  NULL
__tag__receive_time__ string  NULL
Time taken: 0.058 seconds, Fetched 11 row(s)

-- 结果数据源
spark-sql> CREATE TABLE IF NOT EXISTS rds_result
USING jdbc2
OPTIONS (
url="${rdsUrl}",
driver="com.mysql.jdbc.Driver",
dbtable="${rdsTableName}",
user="${user}",
password="${password}",
batchsize="100",
isolationLevel="NONE");

spark-sql> DESC rds_result;
acc double  NULL
label double  NULL
time  string  NULL
Time taken: 0.457 seconds, Fetched 3 row(s)
  • 注册UDF
-- udf_jar_path: 编译完后, emr-sql/target目录下会生成emr-sql_2.11-1.7.0-SNAPSHOT.jar,使用之。

CREATE FUNCTION Logistic_Regression AS 'org.apache.spark.sql.aliyun.udfs.ml.LogisticRegressionUDF' USING JAR '${udf_jar_path}';
  • 提交执行
SET streaming.query.name=lr_prediction;
SET spark.sql.streaming.checkpointLocation.lr_prediction=hdfs:///tmp/spark/lr_prediction;
SET spark.sql.streaming.query.outputMode.lr_prediction=update;
-- 由于DataSource是基于JDBC实现的,所以我们需要设置向RDS表插入数据的SQL
-- 这里我的RDS表名是`result`
SET streaming.query.lr_prediction.sql=insert into `result`(`time`, `label`, `acc`) values(?, ?, ?);

INSERT INTO 
rds_result 
SELECT 
window.start, 
label, 
sum(if(tb.predict = tb.label, 1, 0)) / count(tb.label) as acc 
FROM(
SELECT 
default.Logistic_Regression("${LR_model_path}", concat_ws(" ", label, features)) as predict, 
label, 
__time__ as time 
FROM sls_dataset) tb 
GROUP BY TUMBLING(tb.time, interval 10 second), tb.label;

3.5 效果展示

在DataV中配置上面的RDS结果表,使用折线图查看label=1的预测准确率,如下:
image

4. 小结

本文简要介绍了流式机器学习面临的几个问题,以及相应的解决方法。并使用Spark SQL结合Spark MLlib演示了一个流式机器学习的案例。下一篇,我会简要介绍Spark SQL如何结合阿里云的在线机器学习服务来进行流式机器学习应用开发。

相关实践学习
基于EMR Serverless StarRocks一键玩转世界杯
基于StarRocks构建极速统一OLAP平台
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
目录
相关文章
|
29天前
|
SQL 存储 分布式计算
|
28天前
|
分布式计算 Apache 数据安全/隐私保护
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
35 1
|
11天前
|
Java 前端开发 容器
Struts 2:在技术变革的风暴中航行,探索框架的革命性未来与创新融合之路
【8月更文挑战第31天】Struts 2作为一款成熟的企业级Java Web框架,凭借其稳定性和灵活性自2007年以来广受欢迎。它基于MVC设计模式,支持插件架构和RESTful服务,并能与Spring框架良好集成。然而,面对微服务架构和容器化技术(如Docker和Kubernetes)的兴起,Struts 2需提供更轻量级和支持指南来适应变化。通过深化与现代前端框架(如React和Vue.js)及AI技术的集成,并强化安全性与开发工具,Struts 2有望保持竞争力并迎接未来挑战。
25 0
|
12天前
|
机器学习/深度学习 SQL 数据采集
"解锁机器学习数据预处理新姿势!SQL,你的数据金矿挖掘神器,从清洗到转换,再到特征工程,一网打尽,让数据纯净如金,模型性能飙升!"
【8月更文挑战第31天】在机器学习项目中,数据质量至关重要,而SQL作为数据预处理的强大工具,助力数据科学家高效清洗、转换和分析数据。通过去除重复记录、处理缺失值和异常值,SQL确保数据纯净;利用数据类型转换和字符串操作,SQL重塑数据结构;通过复杂查询生成新特征,SQL提升模型性能。掌握SQL,就如同拥有了开启数据金矿的钥匙,为机器学习项目奠定坚实基础。
22 0
|
20天前
|
SQL 机器学习/深度学习 开发工具
【机器学习 Azure Machine Learning】Azure Machine Learning 访问SQL Server 无法写入问题 (使用微软Python AML Core SDK)
【机器学习 Azure Machine Learning】Azure Machine Learning 访问SQL Server 无法写入问题 (使用微软Python AML Core SDK)
|
25天前
|
SQL 数据挖掘 数据库
SQL计算班级语文平均分:详细步骤与技巧
在数据库管理中,经常需要统计和查询各种汇总信息,如班级某科目的平均分
|
2月前
|
SQL 分布式计算 DataWorks
DataWorks产品使用合集之怎么编写和执行Spark SQL
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
3月前
|
SQL 分布式计算 大数据
MaxCompute产品使用问题之odps sql 底层计算框架是MR吗
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
3月前
|
机器学习/深度学习 调度 云计算
大规模机器学习的计算资源管理
【6月更文挑战第3天】在机器学习中,计算资源是关键所在,相当于驱动模型运行的“燃料”。有效管理计算资源涉及了解硬件性能、合理分配资源及采用优化策略,如任务调度。Python 示例展示了如何使用 multiprocessing 进行并行处理。随着云计算的发展,更多工具帮助我们扩展和管理计算资源。机器学习的计算资源管理是一场持续的探索游戏,旨在实现高效运行和创新成果。准备好投身这个激动人心的领域了吗?
58 1
|
3月前
|
分布式计算 Serverless 数据处理
Serverless Spark计算服务
Serverless Spark计算服务