Я использую pyspark 2.1 для динамического создания разделов из таблицы A в таблицу B. Ниже приведены DDL.
create table A (
objid bigint,
occur_date timestamp)
STORED AS ORC;
create table B (
objid bigint,
occur_date timestamp)
PARTITIONED BY (
occur_date_pt date)
STORED AS ORC;
Затем я использую код pyspark, в котором я пытаюсь определить разделы, которые необходимо объединить, ниже приведена часть кода, где я на самом деле это делаю.
for row in incremental_df.select(partitioned_column).distinct().collect():
path = '/apps/hive/warehouse/B/' + partitioned_column + '=' + format(row[0])
old_df = merge_df.where(col(partitioned_column).isin(format(row[0])))
new_df = incremental_df.where(col(partitioned_column).isin(format(row[0])))
output_df = old_df.subtract(new_df)
output_df = output_df.unionAll(new_df)
output_df.write.option("compression","none").mode("overwrite").format("orc").save(path)
refresh_metadata_sql = 'MSCK REPAIR TABLE ' + table_name
sqlContext.sql(refresh_metadata_sql)
При выполнении кода я могу видеть разделы в HDFS
Найдено 3 элемента -06-01 10:31 /apps/hive/warehouse/B/occur_date_pt=2017-06-02 drwx------ - 307010265 hdfs 0 01-06-2017 10:31 /apps/hive/warehouse/B /occur_date_pt=2017-06-03
Но когда я пытаюсь получить доступ к таблице внутри Spark, я получаю массив за пределами ошибки
>> merge_df = sqlContext.sql('select * from B')
DataFrame[]
>>> merge_df.show()
17/06/01 10:33:13 ERROR Executor: Exception in task 0.0 in stage 200.0 (TID 4827)
java.lang.IndexOutOfBoundsException: toIndex = 3
at java.util.ArrayList.subListRangeCheck(ArrayList.java:1004)
at java.util.ArrayList.subList(ArrayList.java:996)
at org.apache.hadoop.hive.ql.io.orc.RecordReaderFactory.getSchemaOnRead(RecordReaderFactory.java:161)
at org.apache.hadoop.hive.ql.io.orc.RecordReaderFactory.createTreeReader(RecordReaderFactory.java:66)
at org.apache.hadoop.hive.ql.io.orc.RecordReaderImpl.<init>(RecordReaderImpl.java:202)
at org.apache.hadoop.hive.ql.io.orc.ReaderImpl.rowsOptions(ReaderImpl.java:539)
at org.apache.hadoop.hive.ql.io.orc.OrcRawRecordMerger$ReaderPair.<init>(OrcRawRecordMerger.java:183)
at org.apache.hadoop.hive.ql.io.orc.OrcRawRecordMerger$OriginalReaderPair.<init>(OrcRawRecordMerger.java:226)
at org.apache.hadoop.hive.ql.io.orc.OrcRawRecordMerger.<init>(OrcRawRecordMerger.java:437)
at org.apache.hadoop.hive.ql.io.orc.OrcInputFormat.getReader(OrcInputFormat.java:1215)
at org.apache.hadoop.hive.ql.io.orc.OrcInputFormat.getRecordReader(OrcInputFormat.java:1113)
at org.apache.spark.rdd.HadoopRDD$$anon$1.liftedTree1$1(HadoopRDD.scala:252)
at org.apache.spark.rdd.HadoopRDD$$anon$1.<init>(HadoopRDD.scala:251)
at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:211)
at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:102)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
Любая помощь или указатели для решения проблемы будут оценены