文档章节

spark streming写入外部系统

 张欢19933
发布于 2017/06/22 18:25
字数 749
阅读 24
收藏 0
点赞 0
评论 0

SparkStreaming的DStream提供了一个dstream.foreachRDD方法,该方法是一个功能强大的原始的API,它允许将数据发送到外部系统。然而,重要的是要了解如何正确有效地使用这种原始方法。一些常见的错误,以避免如下: 
写数据到外部系统,需要建立一个数据连接对象(例如TCP连接到远程的服务器),使用它将数据发送到外部存储系统。为此开发者可能会在Driver中尝试创建一个连接,然后在worker中使用它来保存记录到外部数据。代码如下:

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()
  }}

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

Spark基于RDD进行编程,RDD的数据不能改变,如果擅长foreachPartition底层的数据可能改变,做到的方式foreachPartition操作一个数据结构,RDD里面一条条数据,但是一条条的记录是可以改变的spark也可以运行在动态数据源上。(就像数组的数据不变,但是指向的索引可以改变) 
我们可以将代码做如下的改进:

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

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

stream.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
  }}

1、需要注意的是: 
(1)、你最好使用forEachPartition函数来遍历RDD,并且在每台Work上面创建数据库的connection。 
(2)、如果你的数据库并发受限,可以通过控制数据的分区来减少并发。 
(3)、在插入MySQL的时候最好使用批量插入。 
(4),确保你写入的数据库过程能够处理失败,因为你插入数据库的过程可能会经过网络,这可能导致数据插入数据库失败。 
(5)、不建议将你的RDD数据写入到MySQL等关系型数据库中。

© 著作权归作者所有

共有 人打赏支持
粉丝 30
博文 422
码字总数 212045
作品 0
海淀
Spark Streaming入门

欢迎大家前往腾讯云+社区,获取更多腾讯海量技术实践干货哦~ 本文将帮助您使用基于HBase的Apache Spark Streaming。Spark Streaming是Spark API核心的一个扩展,支持连续的数据流处理。 什么...

腾讯云加社区
05/16
0
0
Spark中的一些基本概念,及运行流程

1 Spark中的一些基本概念 话不多说,先来一个官网地址:链接 1. Application:用户基于spark的代码,由一个Driver和多个Excutor组成。 2. Application jar:将用户基于spark的代码打包成的jar...

yu0_zhang0
04/23
0
0
Spark的运行架构分析(一)之架构概述

本博客转载自:https://blog.csdn.net/gamer_gyt/article/details/51822765 1:Spark的运行模式 2:Spark中的一些名词解释 3:Spark的运行基本流程 4:RDD的运行基本流程 一:Spark的运行模式...

lubin2016
04/18
0
0
整合Kafka到Spark Streaming——代码示例和挑战

作者Michael G. Noll是瑞士的一位工程师和研究员,效力于Verisign,是Verisign实验室的大规模数据分析基础设施(基础Hadoop)的技术主管。本文,Michael详细的演示了如何将Kafka整合到Spark...

stark_summer
2015/03/03
0
0
Spark 的Core深入(二)

Spark 的 Core 深入(二) 标签(空格分隔): Spark的部分 一、日志清洗的优化: 1.1 日志清洗有脏数据问题 rdd.partitions.length rdd.cacherdd.count 一个分区默认一个task 分区去处理默认...

flyfish225
05/08
0
0
Spark2.x写入Elasticsearch的性能测试

一、Spark集成ElasticSearch的设计动机 ElasticSearch 毫秒级的查询响应时间还是很惊艳的。其优点有: 1. 优秀的全文检索能力 2. 高效的列式存储与查询能力 3. 数据分布式存储(Shard 分片) 相...

openfea
2017/10/27
0
0
Spark基本工作原理与RDD

说在前面的话:由于本人第一次学习spark对知识的理解的深度不够,存在错误的地方还希望大家指出,便于我及时的去改正谢过了哈 一、spark基本的原理 1、分布式 所谓的分布式就客户端在提交spa...

milkcoffeezhu
04/17
0
0
streaming注意事项及调优

部署应用程序 Requirements 运行一个Spark Streaming应用程序,有下面一些步骤 有管理器的集群-这是任何Spark应用程序都需要的需求,详见部署指南 将应用程序打为jar包-你必须编译你的应用程...

hblt-j
2017/10/21
0
0
Spark 学习资源收集【Updating】

(一)spark 相关安装部署、开发环境 1、Spark 伪分布式 & 全分布式 安装指南 http://my.oschina.net/leejun2005/blog/394928 2、Apache Spark探秘:三种分布式部署方式比较 http://dongxic...

大数据之路
2014/09/08
0
1
hadoop和spark的区别你搞明白了吗

学习hadoop已经有很长一段时间了,好像是二三月份的时候朋友给了一个国产Hadoop发行版下载地址,因为还是在学习阶段就下载了一个三节点的学习版玩一下。在研究、学习hadoop的朋友可以去找一下...

左手的倒影
06/27
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

java 复制对象有哪些方式

java 复制对象有哪些方式 Apache的 Common beanutils库 org.apache.commons.beanutils.BeanUtils.copyProperties(dest,origin); Springframework 的BeanUtil 依赖: <dependency> ......

黄威
11分钟前
0
0
1.13 单用户模式

命令 init 6 #重启,reboot或shutdown -r now init 0 #重启,poweroff 单用户模式:密码忘记情况下 主机重启,按方向键↑,选择CentOS Linux (3.10.0-514.e17.X86 64) 7 (Core),按e键编辑...

小丑鱼00
13分钟前
0
0
jstack的简单使用

公司测试反应, 一个java应用的机器,cpu始终是30%, 即使不做交易, 于是想到了之前看到的jstack文章, 实践步骤记录一下: 1, 找出java应用的进程号 ps -ef|grep 应用名|grep -v grep 2, 找出pid...

零二一七
17分钟前
0
0
崛起于Springboot2.X之项目war打包部署(18)

将springboot项目打包步骤: 1、启动类 extends SpringBootServletInitializer 2、启动类添加覆盖方法 @Overrideprotected SpringApplicationBuilder configure(SpringApplicationBuilder......

木九天
26分钟前
0
0
导入CSV文件就行数据整理分析

#-*-coding:utf-8-*-import csv,os,re,mathlocalPath=input("请输入所有群文件的根目录:") #所有QQ群文件的物理根目录路径def info(): info_dic=[] dirList=os.listdi...

Kefy
32分钟前
0
0
CoreText进阶(六)-内容大小计算和自动布局

CoreText进阶(六)-内容大小计算和自动布局 其它文章: CoreText 入门(一)-文本绘制 CoreText入门(二)-绘制图片 CoreText进阶(三)-事件处理 CoreText进阶(四)-文字行数限制和显示更...

aron1992
34分钟前
0
0
一个Unity高人的博客,涉猎范围很广,深度也很深。

https://blog.csdn.net/ecidevilin/article/list/

爽歪歪ES
36分钟前
0
0
Spring Cloud Config-Git后端

EnvironmentRepository的默认实现使用Git后端,这对于管理升级和物理环境以及审核更改非常方便。要更改存储库的位置,可以在Config Server中设置“spring.cloud.config.server.git.uri”配置...

itcloud
38分钟前
1
0
centos7 卸载mysql

[root@zyf ~]# rpm -qa|grep -i mysqlmysql-community-libs-5.6.34-2.el7.x86_64mysql-community-server-5.6.34-2.el7.x86_64mysql-community-release-el7-5.noarchmysql-community-......

Yao--靠自己
46分钟前
0
0
【Spring 系列 条件注解】

Spring 提供了按条件注册Bean的功能涉及到两个组件分别是:核心接口Condition,核心注解Conditional。 1、示例说明 为了演示条件注解的效果,需要定义一个属性文件,然后根据属性文件中配置的...

HansonReal
54分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部