Spark Streaming + Spark SQL 实现配置化ETL流程

Spark Streaming 非常适合ETL。但是其开发模块化程度不高,所以这里提供了一套方案,该方案提供了新的API用于开发Spark Streaming程序,同时也实现了模块化,配置化,并且支持SQL做数据处理。

项目地址

前言

传统的Spark Streaming程序需要:

  • 构建StreamingContext
  • 设置checkpoint
  • 链接数据源
  • 各种transform
  • foreachRDD 输出

通常而言,你可能会因为要走完上面的流程而构建了一个很大的程序,比如一个main方法里上百行代码,虽然在开发小功能上足够便利,但是复用度更方面是不够的,而且不利于协作,所以需要一个更高层的开发包提供支持。

如何开发一个Spark Streaming程序

我只要在配置文件添加如下一个job配置,就可以作为标准的的Spark Streaming 程序提交运行:

{

  "test": {
    "desc": "测试",
    "strategy": "streaming.core.strategy.SparkStreamingStrategy",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "streaming.core.compositor.kafka.MockKafkaStreamingCompositor",
        "params": [
          {
            "metadata.broker.list":"xxx",
            "auto.offset.reset":"largest",
            "topics":"xxx"
          }
        ]
      },
      {
        "name": "streaming.core.compositor.spark.JSONTableCompositor",
        "params": [{"tableName":"test"}
        ]
      },
      {
        "name": "streaming.core.compositor.spark.SQLCompositor",
        "params": [{"sql":"select a from test"}
        ]
      },
      {
        "name": "streaming.core.compositor.RDDPrintOutputCompositor",
        "params": [
          {
          }
        ]
      }
    ],
    "configParams": {
    }
  }
}

上面的配置相当于完成了如下的一个流程:

  1. 从Kafka消费数据
  2. 将Kafka数据转化为表
  3. 通过SQL进行处理
  4. 打印输出

是不是很简单,而且还可以支持热加载,动态添加job等

特性

该实现的特性有:

  1. 配置化
  2. 支持多Job配置
  3. 支持各种数据源模块
  4. 支持通过SQL完成数据处理
  5. 支持多种输出模块

未来可扩展的支持包含:

  1. 动态添加或者删除job更新,而不用重启Spark Streaming
  2. 支持Storm等其他流式引擎
  3. 更好的多job互操作

配置格式说明

该实现完全基于ServiceframeworkDispatcher 完成,核心功能大概只花了三个小时。

这里我们先理出几个概念:

  1. Spark Streaming 定义为一个App
  2. 每个Action定义为一个Job.一个App可以包含多个Job

配置文件结构设计如下:

{

  "job1": {
    "desc": "测试",
    "strategy": "streaming.core.strategy.SparkStreamingStrategy",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "streaming.core.compositor.kafka.MockKafkaStreamingCompositor",
        "params": [
          {
            "metadata.broker.list":"xxx",
            "auto.offset.reset":"largest",
            "topics":"xxx"
          }
        ]
      } ,  
    ],
    "configParams": {
    }
  },
  "job2":{
   ........
 } 
}

一个完整的App 对应一个配置文件。每个顶层配置选项,如job1,job2分别对应一个工作流。他们最终都会运行在一个App上(Spark Streaming实例上)。

  • strategy 用来定义如何组织 compositor,algorithm, ref 的调用关系
  • algorithm作为数据来源
  • compositor 数据处理链路模块。大部分情况我们都是针对该接口进行开发
  • ref 是对其他job的引用。通过配合合适的strategy,我们将多个job组织成一个新的job
  • 每个组件( compositor,algorithm, strategy) 都支持参数配置

上面主要是解析了配置文件的形态,并且ServiceframeworkDispatcher 已经给出了一套接口规范,只要照着实现就行。

模块实现

那对应的模块是如何实现的?本质是将上面的配置文件,通过已经实现的模块,转化为Spark Streaming程序。

以SQLCompositor 的具体实现为例:

class SQLCompositor[T] extends Compositor[T] {

  private var _configParams: util.List[util.Map[Any, Any]] = _
  val logger = Logger.getLogger(classOf[SQLCompositor[T]].getName)

//策略引擎ServiceFrameStrategy 会调用该方法将配置传入进来
  override def initialize(typeFilters: util.List[String], configParams: util.List[util.Map[Any, Any]]): Unit = {
    this._configParams = configParams
  }

// 获取配置的sql语句
  def sql = {
    _configParams(0).get("sql").toString
  }

  def outputTable = {
    _configParams(0).get("outputTable").toString
  }

//执行的主方法,大体是从上一个模块获取SQLContext(已经注册了对应的table),
//然后根据该模块的配置,设置查询语句,最后得到一个新的dataFrame.
// middleResult里的T其实是DStream,我们会传递到下一个模块,Output模块
//params参数则是方便各个模块共享信息,这里我们将对应处理好的函数传递给下一个模块
  override def result(alg: util.List[Processor[T]], ref: util.List[Strategy[T]], middleResult: util.List[T], params: util.Map[Any, Any]): util.List[T] = {
    var dataFrame: DataFrame = null
    val func = params.get("table").asInstanceOf[(RDD[String]) => SQLContext]
    params.put("sql",(rdd:RDD[String])=>{
      val sqlContext = func(rdd)
      dataFrame = sqlContext.sql(sql)
      dataFrame
    })
    middleResult
  }
}

上面的代码就完成了一个SQL模块。那如果我们要完成一个自定义的.map函数呢?可类似下面的实现:

abstract class MapCompositor[T,U] extends Compositor[T]{
  private var _configParams: util.List[util.Map[Any, Any]] = _
  val logger = Logger.getLogger(classOf[SQLCompositor[T]].getName)

  override def initialize(typeFilters: util.List[String], configParams: util.List[util.Map[Any, Any]]): Unit = {
    this._configParams = configParams
  }

  
  override def result(alg: util.List[Processor[T]], ref: util.List[Strategy[T]], middleResult: util.List[T], params: util.Map[Any, Any]): util.List[T] = {
    val dstream = middleResult(0).asInstanceOf[DStream[String]]
    val newDstream = dstream.map(f=>parseLog(f))
    List(newDstream.asInstanceOf[T])
  }
  def parseLog(line:String): U
}

class YourCompositor[T,U] extends MapCompositor[T,U]{

 override def parseLog(line:String):U={
     ....your logical
  }
}

同理你可以实现filter,repartition等其他函数。

总结

该方式提供了一套更为高层的API抽象,用户只要关注具体实现而无需关注Spark的使用。同时也提供了一套配置化系统,方便构建数据处理流程,并且复用原有的模块,支持使用SQL进行数据处理。

广告

这个只是我们大系统的一小部分,愿意和我们一起进一步完善该系统么?欢迎加入我们(请私信我)

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏大数据

Zzreal的大数据笔记-SparkDay04

Spark SQL SparkSQL的前身是Shark,它抛弃原有Shark的代码,汲取了Shark的一些优点,如内存列存储(In-Memory Columna...

1999
来自专栏肖力涛的专栏

Spark踩坑记:共享变量

如果我们想在节点之间共享一份变量,比如一份公共的配置项,该怎么办呢?Spark为我们提供了两种特定的共享变量,来完成节点间变量的共享。 本文首先简单的介绍spa...

1K1
来自专栏CSDN技术头条

Spark Block存储管理分析

Apache Spark中,对Block的查询、存储管理,是通过唯一的Block ID来进行区分的。所以,了解Block ID的生成规则,能够帮助我们了解Blo...

22910
来自专栏岑玉海

Spark调优

因为Spark是内存当中的计算框架,集群中的任何资源都会让它处于瓶颈,CPU、内存、网络带宽。通常,内存足够的情况之下,网络带宽是瓶颈,这时我们就需要进行一些调...

3848
来自专栏Android机器圈

Java设计模式总汇二(小白也要飞)

PS:上一篇我介绍了适配器设计模式、单例设计模式、静态代理设计模式、简单工厂设计模式,如果没有看过第一篇的小火鸡可以点这个看看http://www.cnblog...

3439
来自专栏积累沉淀

Java面试笔试题大汇总(最全+详细答案)

声明:有人说, 有些面试题很变态,个人认为其实是因为我们基础不扎实或者没有深入。本篇文章来自一位很资深的前辈对于最近java面试题目所做的总结归纳,有170道题...

6.4K11
来自专栏24k

Spark wordcount开发并提交到单机(伪分布式)运行

1654
来自专栏Spark生态圈

[spark streaming] ReceiverTracker 数据产生与存储

在Spark Streaming里,总体负责任务的动态调度是JobScheduler,而JobScheduler有两个很重要的成员:JobGenerator 和...

1251
来自专栏分布式系统进阶

Influxdb Cluster下的数据写入

3.2 调用w.MetaClient.CreateShardGroup, 如果ShardGroup存在直接返回ShardGroup信息,如果不存在创建,创建过程...

1532
来自专栏CDA数据分析师

专栏 | Learning Spark (Python版) 学习笔记(二)----键值对、数据读取与保存、共享特性

本来应该上周更新的,结果碰上五一,懒癌发作,就推迟了 = =。以后还是要按时完成任务。废话不多说,第四章-第六章主要讲了三个内容:键值对、数据读取与保存与Spa...

2149

扫码关注云+社区

领取腾讯云代金券