Как мы можем безопасно повторно использовать фрейм данных, читая таблицу Hive, используя HiveWarehouseConnector? - PullRequest
Новая
       6

Как мы можем безопасно повторно использовать фрейм данных, читая таблицу Hive, используя HiveWarehouseConnector?

0 голосов
/ 15 января 2020

При использовании одного и того же кадра данных для чтения таблицы Hive с использованием HiveWarehouseConnector несколько раз во время вычислений возникает исключение.

Пример: 

val hive = com.hortonworks.spark.sql.hive.llap.HiveWarehouseBuilder.session(spark).build()

hive.setDatabase("db")
val df_data = hive.table("table")

val df_one_col = df_data.select("col1")
val df_two_col = df_data.select("col1", "col2")

val df_res = df_two_col.join(df_one_col, "col1")

df_res.show()

Мы получаем ArrayIndexOutOfBoundsException во время выполнения задачи : 

20/01/15 19:46:36 WARN TaskSetManager: Lost task 0.0 in stage 6.0 (TID 18, host, executor 1): java.lang.ArrayIndexOutOfBoundsException: 1
        at org.apache.spark.sql.vectorized.ColumnarBatch.column(ColumnarBatch.java:98)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.datasourcev2scan_nextBatch_0$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
        at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
        at org.apache.spark.scheduler.Task.run(Task.scala:109)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

В строке 39 генерируется код, в котором возникает исключение: 

/* 031 */   ...
/* 032 */   private void datasourcev2scan_nextBatch_0() throws java.io.IOException {
/* 033 */     long getBatchStart = System.nanoTime();
/* 034 */     if (datasourcev2scan_mutableStateArray_0[0].hasNext()) {
/* 035 */       datasourcev2scan_mutableStateArray_1[0] = (org.apache.spark.sql.vectorized.ColumnarBatch)datasourcev2scan_mutableStateArray_0[0].next();
/* 036 */       ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(datasourcev2scan_mutableStateArray_1[0].numRows());
/* 037 */       datasourcev2scan_batchIdx_0 = 0;
/* 038 */       datasourcev2scan_mutableStateArray_2[0] = (org.apache.spark.sql.vectorized.ColumnVector) datasourcev2scan_mutableStateArray_1[0].column(0);
/* 039 */       datasourcev2scan_mutableStateArray_2[1] = (org.apache.spark.sql.vectorized.ColumnVector) datasourcev2scan_mutableStateArray_1[0].column(1);
/* 040 */
/* 041 */     }
/* 042 */     datasourcev2scan_scanTime_0 += System.nanoTime() - getBatchStart;
/* 043 */   }
/* 044 */   ...

ArrayIndexOutOfBoundsException выбрасывается при доступе ко второму столбцу.

Физический план следующий: 

== Physical Plan ==
*(5) Project [col1#333, col2#334]
+- *(5) SortMergeJoin [col1#333], [col1#390], Inner
   :- *(2) Sort [col1#333 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(col1#333, 200)
   :     +- *(1) DataSourceV2Scan [col1#333, col2#334], com.hortonworks.spark.sql.hive.llap.HiveWarehouseDataSourceReader@4527e4
   +- *(4) Sort [col1#390 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(col1#390, 200)
         +- *(3) DataSourceV2Scan [col1#390], com.hortonworks.spark.sql.hive.llap.HiveWarehouseDataSourceReader@4527e4

Мы видим, что мы используем один и тот же экземпляр HiveWarehouseDataSourceReader.

Журналы Spark показывают, что два запущенных запроса Hive запрашивают только 'col1 'column. 

20/01/15 19:46:32 INFO LlapBaseInputFormat: Handle ID c259528d-60ac-42d5-a201-9646335151dd: query=select `col1` from (SELECT * FROM table) as q_d89e3f8df0cd4ce3bdf4c7d938c006ad WHERE col1 IS NOT NULL
...
20/01/15 19:46:35 INFO LlapBaseInputFormat: Handle ID c259528d-60ac-42d5-a201-9646335151dd: query=select `col1` from (SELECT * FROM table) as q_76d9b5c5a5af482a8a3e671b6c8421a8 WHERE col1 IS NOT NULL

Во время оптимизации логического плана Spark сокращение столбца происходило два раза в одном и том же экземпляре HiveWarehouseDataSourceReader, оставляя необходимые столбцы только для "col1". Кажется удивительным, что DataSourceV2Relation, в зависимости от изменяемого читателя, является изменчивым.

Я ищу решение для безопасного повторного использования кадра данных для чтения таблицы Hive с использованием HiveWarehouseConnector.

Я использую HDP 3.1.0 со следующими компонентами:
- Apache Spark 2.3.2
- HiveWarehouseConnector с искровой защелкой 1.0.0
- Hive 3.1.0

1 Ответ

0 голосов
/ 17 января 2020

Spark 2.4
DataSourceV2 был улучшен, особенно SPARK-23203 DataSourceV2 должен использовать неизменяемые деревья

Spark 2.3
Отключить удаление столбцов в считывателе источника данных HiveWarehouseConnector.

Hortonworks уже исправил эту проблему, как указано в Замечаниях по выпуску HDP 3.1.5 .
Мы можем найдите исправление в его репозитории HiveWarehouseConnector github : 

    if (useSpark23xReader) {
      LOG.info("Using reader HiveWarehouseDataSourceReaderForSpark23x with column pruning disabled");
      return new HiveWarehouseDataSourceReaderForSpark23x(params);
    } else if (disablePruningPushdown) {
      LOG.info("Using reader HiveWarehouseDataSourceReader with column pruning and filter pushdown disabled");
      return new HiveWarehouseDataSourceReader(params);
    } else {
      LOG.info("Using reader PrunedFilteredHiveWarehouseDataSourceReader");
      return new PrunedFilteredHiveWarehouseDataSourceReader(params);
    }

Кроме того, интеграция HDP 3.1.5 делает c указать:

Чтобы избежать проблем с правильностью данных в этом выпуске, по умолчанию отключение сокращения и проецирования отключено.
...
Чтобы предотвратить эти проблемы и обеспечить правильные результаты, не включайте сокращение и нажатия.

...