0


基于spark的Hive2Pg数据同步组件

一、背景

    Hive中的数据需要同步到pg供在线使用,通常sqoop具有数据同步的功能,但是sqoop具有一定的问题,比如对数据的切分碰到数据字段存在异常的情况下,数据字段的空值率高、数据字段重复太多,影响sqoop的分区策略,特别是hash分区,调用hash函数容易使得cpu高产生报警**。**同时sqoop的mapreduce任务对数据表的分割以及数据文件也会有一定的不均衡性。为了弥补这些问题,开发了基于spark的数据同步组件,利用spark处理大数据的强大能力及分布式并行性上的优势,通过执行sparksql将数据写入到pg数据库,但是在sparksql 中,保存数据到数据,只有 Append , Overwrite , ErrorIfExists, Ignore 四种模式,不满足特殊场景的需求,尝试利用spark save 源码改进, 批量保存数据,存在则更新不存在则插入。

二、关键设计方案

  1. 方案一

     利用DataFrame框架里自带的df.write.mode(“append”).jdbc(url,pg_table,prop)方法,尝试将df里的每一行row是org.apache.spark.sql.Row类型,结合schema类型转换成DataFrame,方法如下:
    

import org.apache.spark.sql.types._

val map = Map("col1" -> 5, "col2" -> 6, "col3" -> 10)

val (keys, values) = map.toList.sortBy(_._1).unzip

val rows = spark.sparkContext.parallelize(Seq(Row(values: _*)))

val schema = StructType(keys.map(

k => StructField(k, IntegerType, nullable = false)))

val df = spark.createDataFrame(rows, schema)

df.show()

经过分析:转DataFrame的过程有重要的两步:首先是通过spark.sparkContext.parallelize将Row类型转成RDD,其次获取schema后利用spark.createDataFrame****把RDD和schema变为DataFrame。然而为了取到DataFrame的每一行Row,需要调用DataFrame的foreach方法。

Spark的DataFrame的foreach的执行原理:

    Spark DataFrame 的 foreach() 方法将 DataFrame 的每一行作为 Row 对象进行循环,并将给定函数应用于该行。foreach() 的一些限制:Spark中的foreach()方法是在工作节点而不是Driver程序中调用的。这意味着,如果我们在函数内执行print()操作,将无法在会话或笔记本中看到打印结果,因为结果打印在工作节点中。行是只读的,因此您无法更新行的值。鉴于这些限制,foreach() 方法主要用于将每行的一些信息记录到本地计算机或外部数据库foreach方法无法改变原始的DataFrame数据,仅用于迭代处理每个分区的数据。

    foreach方法的处理是并行的,可以提高处理效率,但需要注意处理的顺序可能不同于原始数据的顺序。常规性能调优四:广播大变量默认情况下,task 中的算子中如果使用了外部的变量,每个 task 都会获取一份变量的复 本,这就造成了内存的极大消耗。一方面,如果后续对 RDD 进行持久化,可能就无法将 RDD 数据存入内存,只能写入磁盘,磁盘 IO 将会严重消耗性能;另一方面,task 在创建对象的 时候,也许会发现堆内存无法存放新创建的对象,这就会导致频繁的 GC,GC 会导致工作 线程停止,进而导致 Spark 暂停工作一段时间,严重影响 Spark 性能。假设当前任务配置了 20 个 Executor,指定 500 个 task,有一个 20M 的变量被所有 task 共用,此时会在 500 个 task 中产生 500 个副本,耗费集群 10G 的内存,如果使用了广播变 量, 那么每个 Executor 保存一个副本,一共消耗 400M 内存,内存消耗减少了 5 倍。广播变量在每个 Executor 保存一个副本,此Executor 的所有 task 共用此广播变量,这让变 量产生的副本数量大大减少。在初始阶段,广播变量只在 Driver 中有一份副本。task 在运行的时候,想要使用广播变 量中的数据,此时首先会在自己本地的 Executor 对应的 BlockManager 中尝试获取变量,如 果本地没有,BlockManager 就会从 Driver 或者其他节点的 BlockManager 上远程拉取变量的 复本,并由本地的 BlockManager 进行管理;之后此 Executor 的所有 task 都会直接从本地的 BlockManager 中获取变量。

    一般spark.SparkContext是存在Driver进程里的,工作节点获取不到,每个jvm只能有一个SparkContext,再创建新的SparkContext之前需要先stop()当前活动的SparkContext。

综上分析,方案一不能实现,即DataFrame里不能创建DataFrame

2、方案二

    利用PrepareStatement执行插入更新的sql语句,将DataFrame里的每一行Row的字段数值解析出来封装成sql,每batch个执行一次并提交,碰到有重复的key执行update操作,新的数据执行insert操作。该方案的实现过程借鉴了spark的dataframe的save方法。tableSchema.fields.map(x => x.name).mkString(",")利用map和mkString方法进行字符串操作,同时利用spark的makeSetter方法实现PrepareStatement语句的填充。

三、碰到问题及解决

采用方案二在开发的过程碰到的问题:

  1. Spark error: value foreach is not a member of Object

当调用这样调用的时候:

升级2.12之后,DataFrame的foreachPartition 里面不能处理

Row

的Iterator;

解决方法:(1)

(2)就是使用foreach替代foreachPartition

**2、Caused by: java.io.NotSerializableException:**org.postgreslq.jdbc.PgPrepareStatement

原因: prep是一个PrepareStatement对象,这个对象无法序列化,在标1的地方执行,而传入map中的对象是需要分布式传送到各个节点上,传送前先序列化,到达相应机器上后再反序列化,PrepareStatement是个Java类,如果一个java类想(反)序列化,必须实现Serialize接口,PrepareStatement并没有实现这个接口,对象prep在driver端,collect后的数据也在driver端,就不需prep序列化传到各个节点了。但这样其实会有collect的性能问题。

解决方案:使用mappartition在每一个分区内维持一个pgsql连接进行插入

3、在insert on conflict的语句上报Postgre SQL ERROR:there is no unique or exclusion constraint matching the ON CONFLICT specification。

    执行insert into test values('a','b') on conflict(a,b) do update set c='1';由于建表时没有建关于a,b的CONSTRAINT,于是就会报错,为表添加CONSTRAINT

4、java jar 后面传参数,参数中含有空格的处理方法将含有空格的参数加上双引号。

四、总结

    该组件实现了对于数据规整规范的情况下,直接调用DataFrame的write.mode()方法批量写入,对于特殊的情况hive表里对于pg表的主键字段有重复的情况,进行了重新的封装,通过执行s"INSERT INTO $table ($columns) VALUES ($placeholders) on conflict($id) do update set name1=? ,name2=?"

本文转载自: https://blog.csdn.net/zcc_0015/article/details/135403450
版权归原作者 zcc_0015 所有, 如有侵权,请联系我们删除。

“基于spark的Hive2Pg数据同步组件”的评论:

还没有评论