相关文章推荐
有胆有识的椰子  ·  Pandas ...·  1 周前    · 
飞奔的柚子  ·  Python ...·  1 年前    · 
打盹的酸菜鱼  ·  Malware removes ...·  1 年前    · 
Spark目前支持三种开发语言:Scala、Java、Python,目前我们大量使用Python来开发Spark App(Spark 1.2开始支持使用Python开发Spark Streaming App,我们也准备尝试使用Python开发Spark Streaming App),在这期间关于数据类型的问题曾经困扰我们很长时间,故在此记录一下心路历程。
Spark是使用Scala语言开发的,Hadoop是使用Java语言开发的,Spark兼容Hadoop Writable,而我们使用Python语言开发Spark (Streaming) App,Spark Programming Guides(Spark 1.5.1)其中有一段文字说明了它们相互之间数据类型转换的关系:
也说是说,我们需要处理两个方向的转换:
(1)Writable => Java Type => Python Type;
(2)Python Type => Java Type => Writable;
其中Java Type与Python Type之间数据类型的转换依赖开源组件Pyrolite,相应的数据类型转换如下:
(1)Python Type => Java Type;
(2)Java Type => Python Type;
也就是说,Pyrolite已经为Java Type与Python Type之间的数据类型转换建立了“标准”,我们仅仅需要处理Writable与Java Type之间的数据转换就可以了。
从上图“Writable Support”中可以看出PySpark已经为我们解决了常用的数据类型转换问题,但可以理解为“基本”数据类型,遇到复杂的情况,还是需要我们特殊处理,PySpark已经为我们考虑到了这种业务场景,为我们提供接口Converter(org.apache.spark.api.python.Converter),使得我们可以根据自己的需要扩展数据类型转换机制:
接口Converter仅仅只有一个方法convert,其中T表示源数据类型,U表示目标数据类型,参数obj表示源数据值,返回值表示目标数据值。
Spark Programming Guides(Spark 1.5.1)也为我们举例说明了一个需要自定义Converter的场景:
ArrayWritable是Hadoop Writable的一种,因为Array涉及到元素数据类型的问题,因此使用时需要实现相应的子类,如元素数据类型为整型:
从上面的描述可知,PySpark使用ArrayWritable时涉及到如下两个方向的数据类型转换:
(1)Tuple => Object[] => ArrayWritable;
(2)ArrayWritable => Object[] => Tuple;
我们以IntArrayWritable为例说明如何自定义扩展Converter,同理也需要处理两个方向的数据类型转换:Tuple => Object[] => ArrayWritable、ArrayWritable => Object[] => Tuple。
(1)Tuple => Object[] => IntArrayWritable;
假设我们有一个list,list的元素类型为tuple,而tuple的元素类型为int,我们需要将这个list中的所有数据以SequenceFile的形式保存至HDFS。对于list中的每一个元素tuple,Pyrolite可以帮助我们完成Tuple => Object[]的转换,而Object[] => IntArrayWritable则需要我们自定义Converter实现。
PySpark中使用这个Converter写入数据:
注意:SequenceFile的数据结构为<key, value>,为了简单起见,key指定为com.sina.dip.spark.converter.IntArrayWritable,value指定为org.apache.hadoop.io.NullWritable(即空值)。
运行上述程序时,因为有使用到我们自定义的类,因此需要将com.sina.dip.spark.converter.IntArrayWritable、com.sina.dip.spark.converter.ObjectArrayToIntArrayWritableConverter编译打包为独立的Jar:converter.jar,并通过参数指定,如下:
/usr/lib/spark-1.5.1-bin-2.5.0-cdh5.3.2/bin/spark-submit --jars converter.jar 1.5.1/examples/app/spark_app_save_data_to_seqfile.py
(2)IntArrayWritable => Object[] => Tuple;
我们需要将(1)中写入SequenceFile的Key(IntArrayWritable)还原为list,其中list的元素类型为tuple,tuple的元素类型为int,IntArrayWritable => Object[]也需要用到我们自定义的Converter(Object[] => Tuple由Pyrolite负责):
PySpark使用这个Converter读取数据:
同(1),我们需要将com.sina.dip.spark.converter.IntArrayWritable、com.sina.dip.spark.converter.IntArrayWritableToObjectArrayConverter编译打包为独立的Jar:converter.jar,并通过参数指定,如下:
/usr/lib/spark-1.5.1-bin-2.5.0-cdh5.3.2/bin/spark-submit --jars converter.jar 1.5.1/examples/app/spark_app_read_data_from_seqfile.py
输出结果:
可以看出,通过自定义扩展的Converter:com.sina.dip.spark.converter.ObjectArrayToIntArrayWritableConverter、com.sina.dip.spark.converter.IntArrayWritableToObjectArrayConverter,我们实现了IntArrayWritable(com.sina.dip.spark.converter.IntArrayWritable)与Tuple(Python)之间的转换。