第96课： 通过Spark Streaming的foreachRDD把处理后的数据写入外部存储系统中

本期内容

  • 技术实现解析
  • 实现实战

SparkStreaming的DStream提供了一个dstream.foreachRDD方法，该方法是一个功能强大的原始的API，它允许将数据发送到外部系统。然而,重要的是要了解如何正确有效地使用这种原始方法。一些常见的错误,以避免如下：

写数据到外部系统，需要建立一个数据连接对象（例如TCP连接到远程的服务器），使用它将数据发送到外部存储系统。为此开发者可能会在Driver中尝试创建一个连接，然后在worker中使用它来保存记录到外部数据。例如如下scala代码：

dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { record =>
    connection.send(record) // executed at the worker
  }}

上面的代码是一个错误的演示，因为连接是在Driver中创建的，而写数据是在worker中完成的。此时连接就需要被序列化然后发送到worker中。但是我们知道，连接的信息是不能被序列化和发序列化的（不同的机器连接服务器需要使用不同的服务器端口，即便连接被序列化了也不能使用）

进而我们可以将连接移动到worker中实现，代码如下：

dstream.foreachRDD { rdd =>
  rdd.foreach { record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }}

但是此时，每处理一条数据记录，就需要连接一次外部系统，对于性能来说是个严重的问题。这也不是一个完美的实现。

我们可以将代码做如下的改进：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }}

这样一个partition，只需连接一次外部存储。性能上有大幅度的提高。但是不同的partition之间不能复用连接。我们可以使用连接池的方式，使得partition之间可以共享连接。代码如下：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }}

下面我们使用SparkStreaming实现将数据写到MySQL中：

在pom.xml中加入如下依赖包

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.38</version>
</dependency>
<dependency>
    <groupId>commons-dbcp</groupId>
    <artifactId>commons-dbcp</artifactId>
    <version>1.4</version>
</dependency>

在MySql中创建数据库和表

mysql> create database spark;
Query OK, 1 row affected (0.01 sec)

mysql> use spark;
Database changed
mysql> show tables;
Empty set (0.01 sec)

mysql> create table searchKeyWord(insert_time date,keyword varchar(30),search_count integer);
Query OK, 0 rows affected (0.05 sec)

使用Java编写一个数据库连接池类

package com.dt.spark.common;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

import org.apache.commons.dbcp.BasicDataSource;
import org.apache.log4j.Logger;

/**
 * Description: 数据库连接池类
 * @author dinglq
 */
public class ConnectPool {
    private static Logger log = Logger.getLogger(ConnectPool.class);
    private static BasicDataSource bs = null;

    /**
     * 创建数据源
     * @return
     */
    public static BasicDataSource getDataSource() throws Exception{
        if(bs==null){
            bs = new BasicDataSource();
            bs.setDriverClassName("com.mysql.jdbc.Driver");
            bs.setUrl("jdbc:mysql://spark-master:3306/spark");
            bs.setUsername("root");
            bs.setPassword("vincent");
            bs.setMaxActive(200);//设置最大并发数
            bs.setInitialSize(30);//数据库初始化时，创建的连接个数
            bs.setMinIdle(50);//最小空闲连接数
            bs.setMaxIdle(200);//数据库最大连接数
            bs.setMaxWait(1000);
            bs.setMinEvictableIdleTimeMillis(60*1000);//空闲连接60秒中后释放
            bs.setTimeBetweenEvictionRunsMillis(5*60*1000);//5分钟检测一次是否有死掉的线程
            bs.setTestOnBorrow(true);
        }
        return bs;
    }

    /**
     * 释放数据源
     */
    public static void shutDownDataSource() throws Exception{
        if(bs!=null){
            bs.close();
        }
    }

    /**
     * 获取数据库连接
     * @return
     */
    public static Connection getConnection(){
        Connection con=null;
        try {
            if(bs!=null){
                con=bs.getConnection();
            }else{
                con=getDataSource().getConnection();
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return con;
    }

    /**
     * 关闭连接
     */
    public static void closeCon(ResultSet rs,PreparedStatement ps,Connection con){
        if(rs!=null){
            try {
                rs.close();
            } catch (Exception e) {
                log.error("关闭结果集ResultSet异常！"+e.getMessage(), e);
            }
        }
        if(ps!=null){
            try {
                ps.close();
            } catch (Exception e) {
                log.error("预编译SQL语句对象PreparedStatement关闭异常！"+e.getMessage(), e);
            }
        }
        if(con!=null){
            try {
                con.close();
            } catch (Exception e) {
                log.error("关闭连接对象Connection异常！"+e.getMessage(), e);
            }
        }
    }
}

编写Spark代码：

package com.dt.spark.streaming

import com.dt.spark.common.ConnectPool
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 以网站热词排名为例，将处理结果写到MySQL中
 * Created by dinglq on 2016/5/3.
 */
object WriteDataToMySQL {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("WriteDataToMySQL")
    val ssc = new StreamingContext(conf,Seconds(5))
    // 假设socket输入的数据格式为：searchKeyword,time
    val ItemsStream = ssc.socketTextStream("spark-master",9999)
    // 将输入数据变成(searchKeyword,1)
    var ItemPairs = ItemsStream.map(line =>(line.split(",")(0),1))

     val ItemCount = ItemPairs.reduceByKeyAndWindow((v1:Int,v2:Int)=> v1+v2,Seconds(60),Seconds(10))
    //ssc.checkpoint("/user/checkpoints/")
    // val ItemCount = ItemPairs.reduceByKeyAndWindow((v1:Int,v2:Int)=> v1+v2,(v1:Int,v2:Int)=> v1-v2,Seconds(60),Seconds(10))
    /**
     * 接下来需要对热词的频率进行排序，而DStream没有提供sort的方法。那么我们可以实现transform函数，用RDD的sortByKey实现
     */
    val hottestWord = ItemCount.transform(itemRDD => {
      val top3 = itemRDD.map(pair => (pair._2, pair._1))
        .sortByKey(false).map(pair => (pair._2, pair._1)).take(3)
      ssc.sparkContext.makeRDD(top3)
    })

    hottestWord.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords =>{
        val conn = ConnectPool.getConnection
        conn.setAutoCommit(false);  //设为手动提交
        val  stmt = conn.createStatement();

        partitionOfRecords.foreach( record => {

          stmt.addBatch("insert into searchKeyWord (insert_time,keyword,search_count) values (now(),‘"+record._1+"‘,‘"+record._2+"‘)");
        })

        stmt.executeBatch();
        conn.commit();  //提交事务

      })
    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}

打开netcat发送数据

[email protected]:~# nc -lk 9999
hadoop,1111
spark,2222
spark,3333
hadoop,1111
spark,2222
spark,3333
hadoop,1111
spark,2222
spark,3333
hadoop,1111
spark,2222
spark,3333

运行spark代码

[email protected]:~# /usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.streaming.WriteDataToMySQL  --jars=mysql-connector-java-5.1.38.jar,commons-dbcp-1.4.jar ./spark.jar

查看数据库中的结果：

mysql> select * from searchKeyWord;
+-------------+---------+--------------+
| insert_time | keyword | search_count |
+-------------+---------+--------------+
| 2016-05-03  | spark   |            4 |
| 2016-05-03  | hadoop  |            2 |
| 2016-05-03  | spark   |            4 |
| 2016-05-03  | hadoop  |            2 |
+-------------+---------+--------------+
4 rows in set (0.00 sec)

备注：

1、DT大数据梦工厂微信公众号DT_Spark 
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains

时间： 2024-10-18 12:44:46

第96课： 通过Spark Streaming的foreachRDD把处理后的数据写入外部存储系统中的相关文章

第3课:解读spark –streaming运行机制

感谢DT大数据梦工厂支持提供以下内容,DT大数据梦工厂专注于Spark发行版定制.详细信息请查看 联系邮箱[email protected] 电话:18610086859 QQ:1740415547 微信号:18610086859 定制班:第三课 解读spark –streaming运行机制 一 从实战出发 首先我们运行以下的程序,然后通过这个程序的运行过程进一步加深理解Spark Streaming流处理的Job的执行的过程,代码如下:   def main(args: Array[Strin

第4课：Spark Streaming的Exactly-Once的事务处理和不重复输出彻底掌握

前置知识: 1.事务的特征:1).处理且仅被处理一次:2).输出且只被输出一次 2.SparkStreaming进行事务处理有没有可能处理完全失败? 这个可能性不大,因为Spark是批处理的方式来进行流处理,在SparkStreaming应用程序启动的时候,已经为应用程序分配了相关的资源,而且在调度的过程中可以动态的分配资源,所以除非整个集群所有的硬件都奔溃了,否则一般情况下都会被处理的. 3.SparkStreaming写程序的时候是基于Driver和Executor两部分 SparkStre

第10课：Spark Streaming源码解读之流数据不断接收全生命周期彻底研究和思考

上一课我们讲解了Receiver启动的流程.Receiver是通过ReceiverSupervisor的start方法启动的: /** Start the supervisor */ def start() {   onStart()   startReceiver() } 首先会调用ReceiverSupervisor的onStart()方法, override protected def onStart() {   registeredBlockGenerators.foreach { _.

第89课：Spark Streaming on Kafka解析和安装实战

本课分2部分讲解: 第一部分,讲解Kafka的概念.架构和用例场景: 第二部分,讲解Kafka的安装和实战. 由于时间关系,今天的课程只讲到如何用官网的例子验证Kafka的安装是否成功.后续课程会接着讲解如何集成Spark Streaming和Kafka. 一.Kafka的概念.架构和用例场景 http://kafka.apache.org/documentation.html#introdution 1.Kafka的概念 Apache Kafka是分布式发布-订阅消息系统.它最初由Linked

第4课：Spark Streaming的Exactly-One的事务处理

Spark Streaming的事务处理和关系型数据库的事务的概念有所不同,关系型数据库事务关注的是语句级别的一致性,例如银行转账.而Spark Streaming的事务关注的是某次job执行的一致性.也就是如何保证Job在处理数据的过程中做到如下两点: 不丢失数据 不重复处理数据 SparkStreaming程序执行架构大致如下: 一.我们先来说说丢失数据的情况: Receiver接收到数据后,首先会在Executor级别上保存数据(根据StorageLevel的设置),例如socketTex

第16课：Spark Streaming源码解读之数据清理内幕彻底解密

本期内容: Spark Streaming数据清理原因和现象 Spark Streaming数据清理代码解析 对Spark Streaming解析了这么多课之后,我们越来越能感知,Spark Streaming只是基于Spark Core的一个应用程序,因此掌握Spark Streaming对于我们怎么编写Spark应用是绝对有好处的. Spark Streaming 不像Spark Core的应用程序,Spark Core的应用的数据是存储在底层文件系统,如HDFS等别的存储系统中,而Spar

第4课：Spark Streaming的Exactly-One的事务处理和不重复输出彻底掌握

本篇博文组织形式如下: 一:Exactly-One的事务处理 二:输出不重复 一:Exactly-One的事务处理 一:Exactly-One的事务处理 1. 什么是事务处理: a) 能够处理且只被处理一次.例如,银行转账,A转给B,A有且仅转一次. b) 能够输出,且只能够输出一次.而B接收转账,且直接收一次. 2. 事务处理会不会失败呢? 可能性不大,Spark是批处理的方式来进行流处理Batch Interval的方式,Spark应用程序启动的时候为我们分配了资源,而且在计算的时候也会动态

第4课：Spark Streaming的Exactly Once的事务处理

本期内容: Exactly once 输出不重复 Exactly once 1,事务一定会被处理,且只被处理一次: 2,输出能够输出且只会被输出. Receiver:数据通过BlockManager写入内存+磁盘或者通过WAL来保证数据的安全性. WAL机制:写数据时先通过WAL写入文件系统然后存储的Executor(存储在内存和磁盘中,由StorageLevel设定),假设前面没有写成功后面一定不会存储在Executor,如不存在Executor中的话,汇报Driver数据一定不被处理.WAL

第13课：Spark Streaming源码解读之Driver容错安全性

本期内容: ReceivedBlockTracker容错安全性 DStream和JobGenerator容错安全性 Driver的容错有两个层面:1. Receiver接收数据的元数据 2. Driver管理的各组件信息(调度和驱动层面) 元数据采用了WAL的容错机制 case AddBlock(receivedBlockInfo) =>   if (WriteAheadLogUtils.isBatchingEnabled(ssc.conf, isDriver = true)) {     wa