я проблемы со схемой синхронизации таблиц Hive между Spark и Hive в кластере Mapr с Spark 2.1.0 и Hive 2.1.1.

Мне нужно попытаться решить эту проблему специально для управляемых таблиц, но проблему можно воспроизвести с помощью неуправляемых / внешних таблиц.

Обзор шаговиспользованиеsaveAsTable сохранить кадр данных в заданной таблице.использованиеmode("overwrite").parquet("path/to/table") перезаписать данные для ранее сохраненной таблицы. Я на самом деле изменяю данные через процесс, внешний по отношению к Spark и Hive, но это воспроизводит ту же проблему.использованиеspark.catalog.refreshTable(...) обновить метаданныеЗапросить таблицу сspark.table(...).show(), Любые столбцы, которые были одинаковыми между исходным и перезаписывающим кадрами, будут правильно отображать новые данные, но любые столбцы, которые были только в новой таблице, отображаться не будут.пример

db_name = "test_39d3ec9"
table_name = "overwrite_existing"
table_location = "<spark.sql.warehouse.dir>/{}.db/{}".format(db_name, table_name)

qualified_table = "{}.{}".format(db_name, table_name)
spark.sql("CREATE DATABASE IF NOT EXISTS {}".format(db_name))

Сохранить как управляемую таблицу

existing_df = spark.createDataFrame([(1, 2)])
existing_df.write.mode("overwrite").saveAsTable(table_name)

Обратите внимание, что сохранение в виде неуправляемой таблицы со следующими данными приведет к той же проблеме:

existing_df.write.mode("overwrite") \
    .option("path", table_location) \
    .saveAsTable(qualified_table)

Посмотреть содержимое таблицы

spark.table(table_name).show()
+---+---+
| _1| _2|
+---+---+
|  1|  2|
+---+---+

Перезаписать файлы паркета напрямую

new_df = spark.createDataFrame([(3, 4, 5, 6)], ["_4", "_3", "_2", "_1"])
new_df.write.mode("overwrite").parquet(table_location)

Просматривайте содержимое с помощью программы чтения паркета, содержимое отображается правильно

spark.read.parquet(table_location).show()
+---+---+---+---+
| _4| _3| _2| _1|
+---+---+---+---+
|  3|  4|  5|  6|
+---+---+---+---+

Обновите метаданные spark для таблицы и прочитайте снова как таблицу. Данные будут обновлены для столбцов, которые были одинаковыми, но дополнительные столбцы не отображаются.

spark.catalog.refreshTable(qualified_table)
spark.table(qualified_table).show()
+---+---+
| _1| _2|
+---+---+
|  6|  5|
+---+---+

Я также попытался обновить схему в улье перед вызовомspark.catalog.refreshTable с помощью приведенной ниже команды в оболочке улья:

ALTER TABLE test_39d3ec9.overwrite_existing REPLACE COLUMNS (`_1` bigint, `_2` bigint, `_3` bigint, `_4` bigint);

После запуска команды ALTER я запускаю описать, и он показывает правильно в улье

DESCRIBE test_39d3ec9.overwrite_existing
OK
_1                      bigint
_2                      bigint
_3                      bigint
_4                      bigint

Перед запуском команды alter отображаются только исходные столбцы, как и ожидалось

DESCRIBE test_39d3ec9.overwrite_existing
OK
_1                      bigint
_2                      bigint

Я тогда побежалspark.catalog.refreshTable но это не повлияло на вид искры данных.

Дополнительные примечания

Со стороны искры, я провел большую часть своих испытаний с PySpark, но также тестировал в оболочке spark-shell (scala) и оболочке sparksql. В то время как в искровой оболочке я также пытался использоватьHiveContext но не сработало.

import org.apache.spark.sql.hive.HiveContext
import spark.sqlContext.implicits._
val hiveObj = new HiveContext(sc)
hiveObj.refreshTable("test_39d3ec9.overwrite_existing")

После выполнения команды ALTER в оболочке улья я убедился в Hue, что схема также изменилась там.

Я также попытался запустить команду ALTER сspark.sql("ALTER ...") но версия Spark, на которой мы находимся (2.1.0), не допускает ее, и похоже, что она не будет доступна до Spark 2.2.0 на основе этой проблемы:https://issues.apache.org/jira/browse/SPARK-19261

Я также снова прочитал документацию по искру, особенно этот раздел:https://spark.apache.org/docs/2.1.0/sql-programming-guide.html#hive-metastore-parquet-table-conversion

На основании этих документов,spark.catalog.refreshTable должно сработать. Конфигурация дляspark.sql.hive.convertMetastoreParquet обычноfalse, но я переключил его наtrue для тестирования, и это, похоже, ничего не произвело.

Любая помощь будет оценена, спасибо!