HDP 2.5.0 Sandboxを使用
まず、Scalaでうまくいくか試す。
1) PySpqrkからテストテーブルを作る
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)"nonstrict"を使わないとSparkException: Dynamic partition strict mode requires at least one static partition column.
#sqlContext.setConf("hive.exec.dynamic.partition.mode","nonstrict")
sqlContext.sql("set hive.exec.dynamic.partition.mode=nonstrict")import sqlContext.implicits._
sqlContext.sql("create external table if not exists default.partitiontest1(val string) partitioned by (year int)")
Hiveから確認：
hive> show create table partitiontest1;
OK
CREATE EXTERNAL TABLE `partitiontest1`(
`val` string)
PARTITIONED BY (
`year` int)
ROW FORMAT SERDE
'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
'hdfs://sandbox.hortonworks.com:8020/apps/hive/warehouse/partitiontest1'
TBLPROPERTIES (
'transient_lastDdlTime'='1481536557')
Time taken: 1.071 seconds, Fetched: 14 row(s)
hive>
2) データを入力
import org.apache.spark.sql.SaveMode
Seq(2012 -> "a", 2013 -> "b", 2014 -> "c").toDF("year", "val").write.partitionBy("year") .mode(SaveMode.Append).saveAsTable("default.partitiontest1")
(Sparkと)Hiveから確認：
#sqlContext.sql("show partitions default.partitiontest1").show
hive> select * from partitiontest1;
OK
a 2012
b 2013
c 2014
Time taken: 6.325 seconds, Fetched: 3 row(s)
hive> show partitions partitiontest1;
OK
year=2012
year=2013
year=2014
Time taken: 0.706 seconds, Fetched: 3 row(s)
hive>
PySparkバージョン：
from pyspark.sql import HiveContext
from pyspark.sql import Row
sqlContext = HiveContext(sc)
sqlContext.sql("set hive.exec.dynamic.partition.mode=nonstrict")
...
sqlContext.sql("create external table if not exists default.partitiontest2(val string) partitioned by (year int)")
...
# Using RDD of dict to inferSchema is deprecated. Use pyspark.sql.Row instead
# TypeError: schema should be StructType or list or None
#sc.parallelize([{"2012":"a", "2013":"b","2014":"c"}]).toDF("year", "val").write.partitionBy("year") .mode("append").saveAsTable("default.partitiontest2")
#sqlContext.createDataFrame([{"2012":"a", "2013":"b","2014":"c"}])
http://nadbordrozd.github.io/blog/2016/05/22/one-weird-trick-that-will-fix-your-pyspark-schemas/
from pyspark.sql.types import StringType, StructField, StructType, BooleanType, ArrayType, IntegerType
schema = StructType([StructField("year", IntegerType(), True), StructField("val", StringType(), True)])
# value order matters (eg: year needs to come first) and no label
record = [Row(2012, 'a'), Row(2013, 'b'), Row(2014, 'c')]
#sc.parallelize(record).toDF(schema).collect()
sc.parallelize(record).toDF(schema).write.partitionBy("year") .mode("append").saveAsTable("default.partitiontest2")
sqlContext.sql("show partitions default.partitiontest2").show()
統計情報を更新してみる：
sqlContext.sql("analyze table default.partitiontest1 compute statistics noscan")
# Table is partitioned and partition specification is needed
#sqlContext.sql("analyze table default.partitiontest1 compute statistics")
# ERROR ExecDriver: yarn
# java.lang.LinkageError: ClassCastException: attempting to castjar:file:/usr/hdp/2.5.0.0-1245/spark/lib/spark-assembly-1.6.2.2.5.0.0-1245-hadoop2.7.3.2.5.0.0-1245.jar!/javax/ws/rs/ext/RuntimeDelegate.classtojar:file:/usr/hdp/2.5.0.0-1245/spark/lib/spark-assembly-1.6.2.2.5.0.0-1245-hadoop2.7.3.2.5.0.0-1245.jar!/javax/ws/rs/ext/RuntimeDelegate.class
#sqlContext.sql("analyze table default.partitiontest1 partition(year) compute statistics")
# pyspark.sql.utils.AnalysisException: u"missing KW_STATISTICS at 'for' near '<EOF>'; line 1 pos 61"
#sqlContext.sql("analyze table default.partitiontest1 partition(year) compute for columns")
# Don't see any stats change
# sqlContext.sql("MSCK REPAIR TABLE default.partitiontest2")
Misc.:
sqlContext.sql("select * from default.partitiontest1").collect()
...
[Row(val=u'a', year=2012), Row(val=u'b', year=2013), Row(val=u'c', year=2014)]
>>> sqlContext.sql("select year, val from default.partitiontest1").printSchema()
16/12/13 05:53:41 INFO ParseDriver: Parsing command: select year, val from default.partitiontest1
16/12/13 05:53:41 INFO ParseDriver: Parse Completed
root
|-- year: integer (nullable = true)
|-- val: string (nullable = true)
...
+--------------------+-----+
| key|value|
+--------------------+-----+
|hive.stats.autoga...| true|
+--------------------+-----+
