Spark DataFrame для чтения и записи

Question

У меня есть случай, когда мне нужно загрузить миллионы данных в формате json в таблицы Apache Hive.Поэтому я решил просто загрузить их в датафрейм и записать в виде файлов Parquet.Затем я создам для них внешнюю таблицу.

Я использую Apache Spark 2.1.0 с scala 2.11.8.

Так получилось, что все сообщения следуют своего рода гибкой схеме.Например, столбец «сумма» может иметь значение - 1,0 или 1.

Поскольку я преобразовываю данные из полуструктурированного формата в структурированный формат, но моя схема слегка изменчива, я компенсировал это, полагая параметр inferSchema дляисточники данных, такие как json, помогут мне.

spark.read.option("inferSchema","true").json(RDD[String])

Когда я использовал inferSchema как true при чтении данных json,

case 1: для меньших данных все файлы паркета имеют значение в два раза.

, вариант 2: Для больших данных некоторые файлы паркета имеют значение double, а другие - int64.

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

Мои сомнения / вопросы

1. Когда я пытаюсь вывести схему, не приводит ли она к выводу схемы на полный набор данных?

2. Поскольку я не могу применить какую-либо схему из-за своих ограничений, я решил привести весь столбец к двойному типу данных, поскольку он может содержать как целые, так и десятичные числа.Есть ли более простой способ?

3. Я предполагаю, что поскольку данные разбиты на разделы, inferSchema работает для каждого раздела, а затем выдает общую схему, но не выполняет ничего вроде принудительного применения схемыили что-нибудь в этом роде.Пожалуйста, исправьте меня, если я ошибаюсь.

Примечание: причина, по которой я использую опцию inferSchema, заключается в том, что входящие данные слишком гибкие / переменные, чтобы принудительно реализовать собственный класс дел, хотя некоторыеиз столбцов являются обязательными.Если у вас есть более простое решение, предложите.

Answer 1

На самом деле схема Infer просто обрабатывает все строки, чтобы найти типы. После этого он объединяет результаты, чтобы найти схему, общую для всего набора данных.

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

Чтобы ответить на ваш вопрос, можно вывести схему для вашего ввода. Однако, поскольку вы намереваетесь использовать выходные данные в Hive, вы должны убедиться, что все выходные файлы имеют одинаковую схему.

Простой способ сделать это - использовать кастинг (как вы предлагаете). Как правило, мне нравится делать выбор на заключительном этапе моей работы и просто перечислять все столбцы и типы. Я чувствую, что это делает работу более понятной для человека.

, например

df
.coalesce(numOutputFiles)
.select(
  $"col1"        .cast(IntegerType).as("col1"),
  $"col2"        .cast( StringType).as("col2"),
  $"someOtherCol".cast(IntegerType).as("col3")
)
.write.parquet(outPath)