Реализация real-time загрузки данных в Hive c помощью Kafka topic и Apache Flume

Для сохранения данных с Kafka topic напрямую в Hive можно использовать HiveSink:
Если данные в Kafka топике у Вас сохранены у Вас  в Json-e, то для этого есть .serializer = JSON. Также возможен вариант DELIMITED с последующим указанием разделителя для значений.

Создаём таблицу в Hive для загрузки данных:
CREATE TABLE `db_name.table_name`(
`date` string,`cost` string,`cost_origin` string,`campaign` string, `currency` string)
CLUSTERED BY(campaign)  INTO 10 BUCKETS
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.orc.OrcSerde'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
LOCATION
'/user/flume/flumeingest/db.table_name';


Пример настройки Hive Sink:
tier1.sinks.HiveSink5.channel = channel5
tier1.sinks.HiveSink5.type = hive
tier1.sinks.HiveSink5.hive.metastore = thirft://127.0.0.1:9083
tier1.sinks.HiveSink5.hive.database = db_name
tier1.sinks.HiveSink5.hive.table = table_name
#tier1.sinks.HiveSink5.hive.partition = %{year}
tier1.sinks.HiveSink5.hive.txnsPerBatchAsk = 10
tier1.sinks.HiveSink5.batchSize = 15000
tier1.sinks.HiveSink5.serializer = JSON
#tier1.sinks.HiveSink5.serializer.delimiter = ,
tier1.sinks.HiveSink5.serializer.fieldnames =date,cost,cost_origin,currency,campaign

   На этом можно было бы и остановится, но если Вы хотите использовать  формат таблиц Parquet, то нужно делать доп. телодвижения.
   Проблема в том, что Hive Sink походу пока ещё не умеет сохранять данные в Parquet. При попытке сохранения данных с помощью Hive Sink, возникает следующая ошибка: "parquet.MapredParquetOutputFormat cannot be cast to org.apache.hadoop.hive.ql.io.AcidOutputFormat".
  Возможно проблема кроется в этом баге, который на данный момент ещё не закрыт.
Нам же остаётся только вставлять данные в аналогичную табличку в формате Parquet, с помощью запроса:
INSERT OVERWRITE TABLE `db_name.table_name_parquet` SELECT * FROM `db_name.table_name`;
Тут есть даже пример как автоматизировать этот процесс.

Полезные ссылки:

Коментарі

Дописати коментар

Популярні дописи з цього блогу

Минимальные требование для кластера Hadoop(with Spark)

Зображення
   Есть 4 типы ролей в базовом Hadoop кластере: NameNode (и Standby NameNode), JobTracker,TaskTracker, и DataNode. (node - это машина выполняющая конкретную задачу). Большинство машин в Вашем кластере могут выполнять две из этих ролей, функионируя как DataNode (хранилище данных) и TaskTracker (Для обработки данных).    Следующая диаграмма показывает как должны быть сконфигурированы сервера в зависимости от рабочих нагрузок: Далее приведена рекомендательная спецификация для DataNode/TaskTrackers в Hadoop кластере: 12-24 1-4TB жерстих диска в JBOD (Just a Bunch Of Disks) конфигурации 2 quad-/hex-/octo-core CPUs, 2-2.5GHz 64-512GB ОЗУ Локальная Гигабитная сеть или 10Gigabit сеть (чем большая плотность хранения тем большая пропускная способность необходима) Рекомендации для NameNode/JobTracker/Standby NameNode ноды будет колебаться в зависимости от избыточности: 4–6 1TB жерсткие диски в JBOD конфигурации (1 для OS, 2 для FS [RAID 1], 1 для Apache ZooKee...
Докладніше

Apache Spark Resource Management и модель приложений от YARN

Зображення
Краткий обзор разницы между тем как Spark и MapReduce управляют ресурсами с помощью YARN. Также посмотрим на разницу между Spark и MapReduce архитектурой, почему это важно и как они работают в кластере с ресурс медежрером YARN.  Наиболее популярным YARN приложением после MapReduce является Apache Spark.  Приложения В MapReduce, "job" является блоком наивысшего уровня. Система загружает данные в map функцию, перемешивает их и отправляет в reduce функцию. После записывает результат обратно в хранилище. Spark имеет похожий принцип работы(job может состоять не только из шагов map и reduce), но он имеет ещё один блок уровня управления , который называется "application", который может запустить много job-в в последовательном или паралельном режиме. Архитектура Spark приложений Для тех, кто хорошо знаком с SPARK API, приложению соответствует экземпляр класса SparkContext. В приложении может быть использован однин job, интерактивная сессия с нескольким...
Докладніше

Apache Spark - основные команды

Загрузка данных в DataSet RRD (Resilient Distributed Dataset над которым можно делать преобразования): .parallelize([3,5,6,8,9]) .textFile(path_to_the_source)// указываем хранилище. К примеру hdfs. Трансформация данных : .map(function) — применяет пользовательскую функцию(function) к каждому элементу датасета .filter(function) — возвращает только те элементы датасета, для которых пользовательская функция(function) вернула значение true .distinct([numTasks]) — возвращает датасет, с уникальных элементо первоначального датасета .union(otherDataset).intersection(otherDataset).cartesian(otherDataset) — новый датасет содержит в себе всевозможные пары (A,B), где первый элемент принадлежит исходному датасету, а второй — датасету-аргументу Действия над RDD (Сохранить данные или вывести в консоль..): .saveAsTextFile(path) — сохраняет данные в текстовый файл (в hdfs, на локальную машину или в любую другую поддерживаемую файловую систему — полный список можно посмотреть в докумен...
Докладніше