Spark 2.3.1 AWS EMR не возвращает данные для некоторых столбцов, но работает в Athena / Presto и Spectrum - PullRequest
Новая
       5

Spark 2.3.1 AWS EMR не возвращает данные для некоторых столбцов, но работает в Athena / Presto и Spectrum

0 голосов
/ 13 сентября 2018

Я использую PySpark в Spark 2.3.1 в AWS EMR (Python 2.7.14) 

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL data source example") \
    .config("hive.metastore.client.factory.class", "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory") \
    .config("hive.exec.dynamic.partition", "true") \
    .config("hive.exec.dynamic.partition.mode", "nonstrict") \
    .config("spark.debug.maxToStringFields", 100) \
    .enableHiveSupport() \
    .getOrCreate()


spark.sql('select `message.country` from datalake.leads_notification where `message.country` is not null').show(10)

Это не возвращает данных, 0 строк найдено.Каждое значение для каждой строки в таблице выше возвращается Null.Данные хранятся в PARQUET.

Когда я выполнял один и тот же SQL-запрос на AWS Athena / Presto или на AWs Redshift Spectrum, тогда я получаю все данные столбца, возвращаемые правильно (большинство значений столбца не равны нулю).

Это запрос Athena SQL и Redshift SQL, который возвращает правильные данные: 

select "message.country" from datalake.leads_notification where "message.country" is not null limit 10;

Я использую каталог AWS Glue во всех случаях.Столбец выше НЕ разделен, но таблица разделена на другие столбцы.Я пытался использовать ремонт таблицы, это не помогло.то есть MSCK REPAIR TABLE datalake.leads_notification

Я попытался объединить схему = True, например, так: 

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL data source example") \
    .config("hive.metastore.client.factory.class", "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory") \
    .config("hive.exec.dynamic.partition", "true") \
    .config("spark.sql.parquet.mergeSchema", "true") \
    .config("hive.exec.dynamic.partition.mode", "nonstrict") \
    .config("spark.debug.maxToStringFields", 200) \
    .enableHiveSupport() \
    .getOrCreate()

Без разницы, все же каждое значение одного столбца равно нулю, хотя некоторые не являются нулевыми.

Этот столбец был добавлен как последний столбец таблицы, поэтому большинство данных действительно являются нулевыми, но некоторые строки не являются нулевыми.Наконец, столбец указан в списке столбцов в каталоге, прямо над разделенными столбцами.

Тем не менее Athena / Presto получает все ненулевые значения в порядке, как и Redshift Spectrum, но, увы, EMR Spark 2.3.1 PySpark отображает все значения для этого столбца как «нулевые».Все остальные столбцы в Spark извлекаются правильно.

Может кто-нибудь помочь мне отладить эту проблему, пожалуйста?

Схему куста здесь трудно вырезать и вставить из-за формата вывода. 

***CREATE TABLE datalake.leads_notification(
  message.environment.siteorigin string, 
  dcpheader.dcploaddateutc string, 
  message.id int, 
  message.country string, 
  message.financepackage.id string, 
  message.financepackage.version string)
PARTITIONED BY ( 
  partition_year_utc string, 
  partition_month_utc string, 
  partition_day_utc string, 
  job_run_guid string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://blahblah/leads_notification/leads_notification/'
TBLPROPERTIES (
  'CrawlerSchemaDeserializerVersion'='1.0', 
  'CrawlerSchemaSerializerVersion'='1.0', 
  'UPDATED_BY_CRAWLER'='weekly_datalake_crawler', 
  'averageRecordSize'='3136', 
  'classification'='parquet', 
  'compressionType'='none', 
  'objectCount'='2', 
  'recordCount'='897025', 
  'sizeKey'='1573529662', 
  'spark.sql.create.version'='2.2 or prior', 
  'spark.sql.sources.schema.numPartCols'='4', 
  'spark.sql.sources.schema.numParts'='3', 
  'spark.sql.sources.schema.partCol.0'='partition_year_utc', 
  'spark.sql.sources.schema.partCol.1'='partition_month_utc', 
  'spark.sql.sources.schema.partCol.2'='partition_day_utc', 
  'spark.sql.sources.schema.partCol.3'='job_run_guid', 
  'typeOfData'='file')***

Последние 3 столбца имеют одинаковые проблемы в Spark: 

message.country string, 
message.financepackage.id string, 
message.financepackage.version string

Все возвращают OK в Athena / Presto и Redshift Spectrum, используя один и тот же каталог.

Я прошу прощения за редактирование.

спасибо

Ответы [ 2 ]

0 голосов
/ 14 сентября 2018 
spark = SparkSession \
        .builder \
        .appName("Python Spark SQL data source example") \
        .config("hive.metastore.client.factory.class", "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory") \
        .config("hive.exec.dynamic.partition", "true") \
        .config("spark.sql.parquet.mergeSchema", "true") \
        .config("spark.sql.hive.convertMetastoreParquet", "false") \
        .config("hive.exec.dynamic.partition.mode", "nonstrict") \
        .config("spark.debug.maxToStringFields", 200) \
        .enableHiveSupport() \
        .getOrCreate()

Это решение: обратите внимание на 

 .config("spark.sql.hive.convertMetastoreParquet", "false")

Все столбцы схемы в нижнем регистре, и схема была создана AWS Glue, а не моим пользовательским кодом, поэтому я не знаю, что именновызвало проблему, поэтому использование вышеуказанного, вероятно, является безопасной настройкой по умолчанию, когда создание схемы не находится под вашим непосредственным контролем.Это главная ловушка, ИМХО, поэтому я надеюсь, что это поможет кому-то еще в будущем.Спасибо tooptoop4 , которые указали на статью:

https://medium.com/@an_chee/why-using-mixed-case-field-names-in-hive-spark-sql-is-a-bad-idea-95da8b6ec1e0

0 голосов
/ 13 сентября 2018

сделать шаг 5 проверка схемы: http://www.openkb.info/2015/02/how-to-build-and-use-parquet-tools-to.html

Моя ставка заключается в том, что эти новые имена столбцов в определении паркета либо в верхнем регистре (в то время как другие имена столбцов в нижнем регистре), либо имена новых столбцов в определении паркета либострочные буквы (в то время как имена других столбцов прописные)

см. Проблемы с искрой при чтении файлов паркета https://medium.com/@an_chee/why-using-mixed-case-field-names-in-hive-spark-sql-is-a-bad-idea-95da8b6ec1e0

...