Как загрузить данные в hadoop python

Process Mining на Hadoop

В данной статье будет рассказано, как можно на Hadoop проанализировать и обработать лог, рассчитать граф и отрисовать его на локальном компьютере, используя Graphviz.

Для начала нужно загрузить данные на HDFS. Самый простой способ — копирование файлов через веб-интерфейс файлового менеджера в Hue. Веб-интерфейс интуитивно понятен и расположен по адресу http://[Hue_node]:8888/filebrowser/. Однако, если размер файла от нескольких гигабайт, то таким способом данные скорее всего могут не загрузиться. В таком случае можно рекомендовать использовать WinSCP.

Когда файлы будут загружены с помощью WinSCP, необходимо выполнить команду для их перемещения в HDFS:

hadoop fs -put /путь_куда_загружен_файл /путь_куда_переместить_файл

Если планируется загружать данные в таблицы и работать с HiveQL запросами, то каждый файл при перемещении на HDFS нужно размещать в отдельной директории.

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

create schema bdp location 'hdfs://pathdirectory'; use bdp; create external table if not exists bdp.logfull (TRANSACTIONLOGNO bigint, USERNAME string, TRANSACTIONDATE timestamp, SESSIONNO bigint, BANKACCOUNTNO bigint, AUDITMESSAGE string, CUSTOMERNO bigint, APPLICATIONNO bigint, BANKPRODUCTNO bigint, MODBANKACCOUNTNO bigint, BANKACCOUNTROLENO bigint, AUTHENTICATIONNO bigint) row format delimited fields terminated by ';' stored as textfile location 'hdfs://pathdirectory' tblproperties("serialization.encoding"="windows-1251", "skip.header.line.count"="1", "transactional"="true"); select * from bdp.logfull limit 1000; select count(*) from bdp.logfull; create external table if not exists bdp.logfileclear (TRANSACTIONLOGNO bigint, USERNAME string, TRANSACTIONDATE timestamp, AUDITMESSAGE string, CUSTOMERNO bigint, APPLICATIONNO bigint, BANKPRODUCTNO bigint) insert into bdp.logfileclear select TRANSACTIONLOGNO, USERNAME, TRANSACTIONDATE, AUDITMESSAGE, CUSTOMERNO, APPLICATIONNO, BANKPRODUCTNO from bdp.logfull where (username!='' and username not like '%admin%' and username not like '%importAdmin%') and (auditmessage!=''); insert overwright local directory '/pathdirectory' row format delimited fields terminated by ';' select * from bdp.logfileclear

Когда лог проанализирован и очищен, можно приступать к расчету графа. На данном этапе можно использовать PySpark. Код для подключения в JupyterHub к PySpark приведен ниже:

import os import sys spark_home='/opt/cloudera/parcels/SPARK2/lib/spark2' os.environ['SPARK_HOME']=spark_home os.environ['PYSPARK_DRIVER_PYTHON']='/opt/venvs/anaconda/bin/python' os.environ['PYSPARK_PYTHON']='/opt/venv/anaconda/bin/python' sys.path.insert(0, os.path.join (spark_home, 'python')) sys.path.insert(0, os.path.join (spark_home, 'python/lib/py4j-0.10.7-src.zip')) from pyspark import SparkContext,SparkConf, HiveContext from pyspark.sql import SparkSession, SQLContext from pyspark.sql.types import * conf=SparkConf().setAppName('test').\ setMaster("yarn-client").\ set('spark.local.dir', 'sparktmp').\ set('spark.executor.memory', '3g').\ set('spark.driver.maxResultSize', '5g').\ set('spark.executor.instances', '6').\ set('spark.port.maxRetries', '150').\ set('spark.executor.cores', '10').\ set('spark.dynamicAllocation.enabled', 'false') sc=SparkContext.getOrCreate(conf=conf) sc 

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

!ls -la /opt/cloudera/parcels/ !ls -la /opt/cloudera/parcels/SPARK2/lib/spark2/python/lib 

Данные из файла загружаем в RDD.

RDD — основная концепция Spark

Ключ к пониманию Spark — это RDD: Resilient Distributed Dataset. По сути это надежная распределенная таблица (на самом деле RDD содержит произвольную коллекцию, но удобнее всего работать с кортежами, как в реляционной таблице). RDD может быть полностью виртуальной и просто знать, как она породилась, чтобы, например, в случае сбоя узла, восстановиться. А может быть и материализована — распределенно, в памяти или на диске (или в памяти с вытеснением на диск). Также, внутри, RDD разбита на партиции — это минимальный объем RDD, который будет обработан каждым рабочим узлом.

Перед расчетом графа нужно определить какие столбцы являются time:timestamp, case:concept:name, case:concept, с помощью PySpark дату нужно преобразовать к необходимому формату YYYY-MM-DD HH:MI:SS, отсортировать лог по времени. Затем лог передать в PM4PY и произвести расчет с помощью выбранного майнера.

И в завершении статьи небольшой лайфхак �� Предположим, на Hadoop отсутствует возможность установить Graphviz и отрисовать рассчитанный граф можно только на локальном компьютере. Есть простое решение! Рассчитанный граф сохраняется в файл pickle, скачивается на локальный компьютер и затем отрисовывается. Расчет графа и сохранение в pickle-файл:

from pm4py.visualization.heuristics_net import factory as hn_vis_factory from pm4py.objects.heuristics_net.net import HeuristicsNet import pickle def save_object(obj, filename): with open(filename, 'wb') as input: pickle.dump(obj, output, pickle.HIGHEST_PROTOCOL) final_net=HeuristicsNet(result[1]) final_net.calculate() save_object(final_net, 'result_pkl.pkl') sc.stop() 

«Отрисовка» графа на компьютере:

 with open('result_pkl.pkl', 'rb') as input: final_net=pickle.load(input) gviz=hn_vis_factory.apply(final_net) hn_vis_factory.view(gviz) 

Итак, если у Вас большой лог-файл и необходимо построить граф процесса, то вы не ограничены возможностями вашего компьютера и в обработке больших данных вам может помочь Hadoop.

Обработка больших данных: первые шаги в понимании Hadoop MapReduce и Spark

Big Data как концепт довольно понятна, но из-за того, что она включает в себя множество процессов, сложно сказать, с чего именно нужно начать изучение. Как хранятся файлы? Или как получать эти файлы? А может, сразу — как анализировать данные? О своём опыте работы с Big Data и почему Spark лучше, чем Hadoop MapReduce в обработке данных, рассказывает Эмилия Межекова, ETL-developer в Luxoft.

Выше показана экосистема больших данных и примеры инструментов, которые можно использовать для каждой группы. Выглядит устрашающе, но нам нужно разобраться лишь в том, как именно данные обрабатываются, — вернее, рассмотреть два варианта, как это можно сделать с помощью следующих фреймворков: Hadoop MapReduce и Apache Spark.

Hadoop MapReduce и что его окружает

Apache Hadoop — инфраструктура, упрощающая работу с кластерами.

Основные элементы Hadoop — это:

• распределённая файловая система (HDFS);
• метод крупномасштабного выполнения программ (MapReduce).

HDFS — распределённая файловая система Hadoop для хранения файлов больших размеров с возможностью потокового доступа к информации, поблочно распределённой по узлам вычислительного кластера. Здесь мы храним, читаем, записываем и перекладываем данные.

MapReduce — модель распределённых вычислений, представленная компанией Google, используемая для параллельных вычислений над очень большими, вплоть до нескольких петабайт, наборами данных в компьютерных кластерах.

Алгоритм легко понять по аналогии

Представьте, что вам предложено подсчитать голоса на национальных выборах. В вашей стране 25 партий, 2500 избирательных участков и 2 миллиона граждан. Как это можно сделать? Можно собрать все избирательные бюллетени со всех участков и подсчитать их самостоятельно, либо приказать каждому избирательному участку подсчитать голосов по каждой из 25 партий и передать вам результат, после чего объединить их по партиям.

Ниже представлена схема выполнения данного алгоритма на примере подсчёта слов в выборке.

Разберём, что происходит, по этапам;

• Input — входные данные для обработки;
• Splitting — разбивка данных на порционные данные;
• Mapping — обработка этих порционных данных воркерами (вычислительными процессами) в формате ключ-значение. Для этого алгоритма ключ — слово, значение — количество вхождений данного слова;
• Shuffling — ключи сортируются, чтобы упростить обобщение данных и сделать всю работу в одном воркере, не раскидывая их по разным местам;
• Reducing — после того, как мы посчитали количество одинаковых слов на каждом отдельном воркере, объединяем их вместе.

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

Но есть и недостатки, обусловленные архитектурными особенностями этой вычислительной модели:

• недостаточно высокая производительность: классическая технология, в частности, реализованная в ядре Apache Hadoop, обрабатывает данные ациклично в пакетном режиме. При этом функции Reduce не запустятся до завершения всех процессов Map. Все операции проходят по циклу чтение-запись с жёсткого диска, что влечёт задержки в обработке информации;
• ограниченность применения: высокие задержки распределённых вычислений, приемлемые в пакетном режиме обработки, не позволяют использовать классический MapReduce для потоковой обработки в режиме реального времени повторяющихся запросов и итеративных алгоритмов на одном и том же датасете, как в задачах машинного обучения. Для решения этой проблемы, свойственной Apache Hadoop, были созданы другие Big Data — фреймворки, в частности Apache Spark;
• программисту необходимо прописывать код для этапов Map и Reduce самостоятельно.

Apache Spark

В своей работе мне приходится очень часто писать SQL-запросы и смотреть, какие данные приходят на вход и что внутри них хранится. Для этих целей мне хочется, чтобы инструмент был более интерактивным и не приходилось ждать выполнения запроса часами (но скорость зависит от количества данных, естественно). В этом поможет Spark, он работает намного быстрее Hadoop MapReduce.

Spark — инфраструктура кластерных вычислений, сходная с Hadoop MapReduce. Однако Spark не занимается ни хранением файлов в файловой системе, ни управлением ресурсами. Spark обрабатывает данные ещё быстрее с помощью встроенных коллекций RDD (Resilient Distributed Datasets), которые дают возможность выполнять вычисления в больших кластерах. Благодаря RDD можно совершать такие операции, как map, join, reduce, записывать данные на диск и загружать их.

Добавлю таблицу для сравнения Hadoop MapReduce и Spark.

Но как же достигается данное ускорение? Ниже представлены самые значимые решения в архитектуре Spark.

• Промежуточные данные вычислений не записываются на диск, а образуют своего рода общую оперативную память. Это позволяет разным рабочим процессам использовать общие переменные и их состояния.
• Отложенные вычисления: Spark приступает к выполнению запроса лишь при непосредственном обращении к нему (вывод на экран, запись конечных данных на диск). В этом случае срабатывает планировщик, соединяя все преобразования, написанные ранее.

Из-за некоторых архитектурных особенностей Hadoop MapReduce уступает по скорости Spark. Для своих задач я выбрала Spark, потому что при моём наборе данных и итерациях он работает быстрее. Мне было интересно посмотреть, что было до инструмента, которым я пользуюсь, и каким образом всё развивалось. Это лишь общее описание работы этих фреймворков, дающее немного понять, как всё внутри обрабатывается. Зная, как работает тот и другой алгоритм, вы теперь можете выбрать для себя подходящий.

Использование определяемых пользователем функций Python с Apache Hive и Apache Pig в HDInsight

Узнайте, как использовать определяемые пользователем функции (UDF) Python с Apache Hive и Apache Pig в Apache Hadoop на кластерах Azure HDInsight.

Python в HDInsight

Python2.7 устанавливается по умолчанию в HDInsight 3.0 и более поздних версий. Apache Hive можно использовать с этой версией Python для потоковой обработки. При этом для передачи данных между Hive и определяемой пользователем функцией используется STDOUT и STDIN.

В состав HDInsight также входят Jython, который представляет собой реализацию Python, написанную на Java. Jython выполняется непосредственно на виртуальной машине Java и не использует потоковую передачу. Jython является рекомендуемым интерпретатором Python при использовании Python с Pig.

Предварительные требования

• Кластер Hadoop в HDInsight. Ознакомьтесь со статьей Краткое руководство. Использование Apache Hadoop и Apache Hive в Azure HDInsight с шаблоном Resource Manager.
• Клиент SSH. Дополнительные сведения см. в руководстве по подключению к HDInsight (Apache Hadoop) с помощью SSH.
• Схема универсального кода ресурса (URI) для основного хранилища кластеров. wasb:// для службы хранилища Azure, abfs:// для Azure Data Lake Storage 2-го поколения или adl:// для Azure Data Lake Storage 1-го поколения. Если для службы хранилища Azure включено безопасное перемещение, URI будет таким: wasbs://. См. также сведения о безопасной передаче.
• Возможное изменение конфигурации хранилища. Если используется учетная запись хранения типа BlobStorage , см. раздел Конфигурация хранилища.
• Необязательный элемент. Если вы планируете использовать PowerShell, необходимо установить модуль AZ .

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

Конфигурация хранилища

Если используемая учетная запись хранения относится к типу Storage (general purpose v1) или StorageV2 (general purpose v2) , ничего делать не нужно. Процесс, приведенный в этой статье, создает выходные данные по крайней мере /tezstaging для . Конфигурация hadoop по умолчанию содержится /tezstaging в переменной fs.azure.page.blob.dir конфигурации в для core-site.xml службы HDFS . Эта конфигурация приводит к выводу в каталог страничных BLOB-объектов, которые не поддерживаются для учетной записи хранения. BlobStorage Чтобы использовать BlobStorage для процесса, описанного в этой статье, удалите /tezstaging из переменной конфигурации fs.azure.page.blob.dir . Для доступа к конфигурации используйте пользовательский интерфейс Ambari. В противном случае вы получите сообщение об ошибке: Page blob is not supported for this account type.

Шаги в этом документе основаны на следующих предположениях:

• Вы создаете скрипты Python в локальной среде разработки.
• Вы отправляете скрипты в HDInsight, используя команду scp или предоставленный скрипт PowerShell.

Если для работы с HDInsight вы хотите использовать Azure Cloud Shell (bash), вам необходимо:

• Создать скрипты в среде Cloud Shell.
• Использовать scp для отправки файлов из Cloud Shell в HDInsight.
• Использовать ssh из Cloud Shell для подключения к HDInsight и выполнения примеров.

Определяемая пользователем функция Apache Hive

Скрипт Python можно использовать в качестве определяемой пользователем функции из Hive через HiveQL с помощью инструкции TRANSFORM . Например, следующий запрос HiveQL вызывает файл hiveudf.py , хранящийся в учетной записи хранения Azure по умолчанию для кластера.

add file wasbs:///hiveudf.py; SELECT TRANSFORM (clientid, devicemake, devicemodel) USING 'python hiveudf.py' AS (clientid string, phoneLabel string, phoneHash string) FROM hivesampletable ORDER BY clientid LIMIT 50; 

Вот что делает данный пример:

1. Инструкция add file в начале файла добавляет файл hiveudf.py в распределенный кэш, и он становится доступен всем узлам кластера.
2. Инструкция SELECT TRANSFORM . USING выбирает данные из hivesampletable . Она также передает параметры clientid, devicemake и devicemodel в скрипт hiveudf.py .
3. Предложение AS описывает поля, возвращаемые из hiveudf.py .

Создать файл

В среде разработки создайте текстовый файл с именем hiveudf.py . Используйте следующий код в качестве содержимого файла:

#!/usr/bin/env python import sys import string import hashlib while True: line = sys.stdin.readline() if not line: break line = string.strip(line, "\n ") clientid, devicemake, devicemodel = string.split(line, "\t") phone_label = devicemake + ' ' + devicemodel print "\t".join([clientid, phone_label, hashlib.md5(phone_label).hexdigest()]) 

Сценарий выполняет следующие действия:

1. Считывается строка данных из STDIN.
2. Стоящий в конце знак новой строки удаляется с помощью string.strip(line, «\n «) .
3. При обработке потока в одной строке будут содержаться все значения, разделенные символом табуляции. Поэтому можно использовать string.split(line, «\t») для разделения входящих данных при каждой табуляции, возвращая лишь поля.
4. По завершении обработки результат должен быть записан в поток STDOUT в виде одной строки, с разделенными символами табуляции полями. Например, print «\t».join([clientid, phone_label, hashlib.md5(phone_label).hexdigest()]) .
5. Цикл while повторяется до тех пор, пока считывается line .

Выходные данные скрипта представляют собой объединенные входные значения для devicemake и devicemodel , а также хэш для объединенного значения.

Отправка файла (оболочка)

Следующая команда заменяет sshuser фактическим именем пользователя, если оно отличается. Вместо mycluster укажите реальное имя кластера. Рабочей должна быть папка, в которой находится файл.

Используйте scp для копирования файлов в кластер HDInsight. Измените и введите команду :

scp hiveudf.py sshuser@mycluster-ssh.azurehdinsight.net: 

ssh sshuser@mycluster-ssh.azurehdinsight.net 

hdfs dfs -put hiveudf.py /hiveudf.py 

Использование Hive UDF (оболочка)

1. Чтобы подключиться к Hive, в открытом сеансе SSH введите следующую команду:

beeline -u 'jdbc:hive2://headnodehost:10001/;transportMode=http' 

add file wasbs:///hiveudf.py; SELECT TRANSFORM (clientid, devicemake, devicemodel) USING 'python hiveudf.py' AS (clientid string, phoneLabel string, phoneHash string) FROM hivesampletable ORDER BY clientid LIMIT 50; 

100041 RIM 9650 d476f3687700442549a83fac4560c51c 100041 RIM 9650 d476f3687700442549a83fac4560c51c 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 

Отправка файла (PowerShell)

PowerShell также можно использовать для удаленного запуска запросов на использование Hive. Рабочей должна быть папка, в которой находится hiveudf.py . Чтобы выполнить запрос Hive, использующий скрипт hiveudf.py , примените следующий скрипт PowerShell:

# Login to your Azure subscription # Is there an active Azure subscription? $sub = Get-AzSubscription -ErrorAction SilentlyContinue if(-not($sub)) < Connect-AzAccount ># If you have multiple subscriptions, set the one to use # Select-AzSubscription -SubscriptionId "" # Revise file path as needed $pathToStreamingFile = ".\hiveudf.py" # Get cluster info $clusterName = Read-Host -Prompt "Enter the HDInsight cluster name" $clusterInfo = Get-AzHDInsightCluster -ClusterName $clusterName $resourceGroup = $clusterInfo.ResourceGroup $storageAccountName=$clusterInfo.DefaultStorageAccount.split('.')[0] $container=$clusterInfo.DefaultStorageContainer $storageAccountKey=(Get-AzStorageAccountKey ` -ResourceGroupName $resourceGroup ` -Name $storageAccountName)[0].Value # Create an Azure Storage context $context = New-AzStorageContext ` -StorageAccountName $storageAccountName ` -StorageAccountKey $storageAccountKey # Upload local files to an Azure Storage blob Set-AzStorageBlobContent ` -File $pathToStreamingFile ` -Blob "hiveudf.py" ` -Container $container ` -Context $context 

Дополнительные сведения об отправке файлов см. в статье Отправка данных для заданий Apache Hadoop в HDInsight.

Использование Hive UDF

# Script should stop on failures $ErrorActionPreference = "Stop" # Login to your Azure subscription # Is there an active Azure subscription? $sub = Get-AzSubscription -ErrorAction SilentlyContinue if(-not($sub)) < Connect-AzAccount ># If you have multiple subscriptions, set the one to use # Select-AzSubscription -SubscriptionId "" # Get cluster info $clusterName = Read-Host -Prompt "Enter the HDInsight cluster name" $creds=Get-Credential -UserName "admin" -Message "Enter the login for the cluster" $HiveQuery = "add file wasbs:///hiveudf.py;" + "SELECT TRANSFORM (clientid, devicemake, devicemodel) " + "USING 'python hiveudf.py' AS " + "(clientid string, phoneLabel string, phoneHash string) " + "FROM hivesampletable " + "ORDER BY clientid LIMIT 50;" # Create Hive job object $jobDefinition = New-AzHDInsightHiveJobDefinition ` -Query $HiveQuery # For status bar updates $activity="Hive query" # Progress bar (optional) Write-Progress -Activity $activity -Status "Starting query. " # Start defined Azure HDInsight job on specified cluster. $job = Start-AzHDInsightJob ` -ClusterName $clusterName ` -JobDefinition $jobDefinition ` -HttpCredential $creds # Progress bar (optional) Write-Progress -Activity $activity -Status "Waiting on query to complete. " # Wait for completion or failure of specified job Wait-AzHDInsightJob ` -JobId $job.JobId ` -ClusterName $clusterName ` -HttpCredential $creds # Uncomment the following to see stderr output # Progress bar (optional) Write-Progress -Activity $activity -Status "Retrieving output. " # Gets the log output Get-AzHDInsightJobOutput ` -Clustername $clusterName ` -JobId $job.JobId ` -HttpCredential $creds 

Результат выполнения задания Hive должен выглядеть аналогично следующему примеру:

100041 RIM 9650 d476f3687700442549a83fac4560c51c 100041 RIM 9650 d476f3687700442549a83fac4560c51c 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 100042 Apple iPhone 4.2.x 375ad9a0ddc4351536804f1d5d0ea9b9 

Определяемая пользователем функция Apache Pig

Скрипт Python можно использовать в виде определяемой пользователем функции из Pig с использованием инструкции GENERATE . Вы можете запустить скрипт с помощью Jython или CPython.

• Jython работает на виртуальной машине Java и изначально может вызываться из Pig.
• CPython является внешним процессом, поэтому данные из Pig на JVM отправляются в скрипт, выполняющийся в процессе Python. Выходные данные скрипта Python отправляются обратно в Pig.

Чтобы указать интерпретатор Python, используйте register при указании ссылки на скрипт Python. Следующие примеры регистрируют скрипты с Pig в качестве myfuncs :

• Для использования Jython: register ‘/path/to/pigudf.py’ using jython as myfuncs;
• Для использования CPython: register ‘/path/to/pigudf.py’ using streaming_python as myfuncs;

При использовании Jython путь к файлу pig_jython может быть локальным или указан как WASBS://. Но при использовании CPython необходимо указать ссылку на файл в локальной файловой системе узла, который используется для отправки задания Pig.

После регистрации язык Pig Latin будет одинаковым для обоих примеров:

LOGS = LOAD 'wasbs:///example/data/sample.log' as (LINE:chararray); LOG = FILTER LOGS by LINE is not null; DETAILS = FOREACH LOG GENERATE myfuncs.create_structure(LINE); DUMP DETAILS; 

Вот что делает данный пример:

1. Первая строка загружает образец файла данных sample.log в LOGS . Она также определяет каждую запись как массив символов chararray .
2. Следующая строка отфильтровывает все пустые значения, сохраняя результат работы в LOG .
3. Затем выполняется итерация по записям в LOG и используется инструкция GENERATE для вызова метода create_structure , содержащегося в скрипте Python или Jython, загруженном как myfuncs . LINE используется для передачи текущей записи в функцию.
4. Наконец, выходные данные сбрасываются в поток STDOUT командой DUMP . После завершения операции эта команда выведет результат.

Создать файл

В среде разработки создайте текстовый файл с именем pigudf.py . Используйте следующий код в качестве содержимого файла:

# Uncomment the following if using C Python #from pig_util import outputSchema @outputSchema("log: <(date:chararray, time:chararray, classname:chararray, level:chararray, detail:chararray)>") def create_structure(input): if (input.startswith('java.lang.Exception')): input = input[21:len(input)] + ' - java.lang.Exception' date, time, classname, level, detail = input.split(' ', 4) return date, time, classname, level, detail 

В примере с Pig Latin входные данные LINE определены как массив знаков, т. к. согласованной схемы входных данных нет. Скрипт Python выполняет преобразование данных в согласованную схему на выходе.

1. Инструкция @outputSchema задает формат данных, в котором они возвращаются в Pig. В данном случае это data bag, являющийся типом данных Pig. Корзина содержит следующие поля, все они имеют тип «Массив строк» (строки):
   • date — дата создания записи журнала;
   • date — время создания записи журнала;
   • classname — имя класса, для которого создана запись;
   • level — уровень журналирования;
   • detail — подробная информация о записи журнала.
2. Затем def create_structure(input) определяет функцию, в которую Pig отправляет строковые элементы.
3. Данные для примера, sample.log , в основном соответствуют схеме даты, времени, имени класса, уровня и подробной информации. Однако он содержит несколько строк, начинающихся с *java.lang.Exception* . Эти строки должны быть изменены в соответствии со схемой. Инструкция if проверяет на наличие таких строк, затем обрабатывает входные данные, переставляя строку *java.lang.Exception* в конец, формируя данные в соответствии с ожидаемой схемой.
4. Затем команда split используется для разделения данных по первым четырем символам пробела. Выходным данным присваиваются значения date , time , classname , level и detail .
5. И результаты возвращаются в Pig.

Когда данные возвращаются в Pig, они имеют согласованную схему, определенную инструкцией @outputSchema .

Отправка файла (оболочка)

В командах ниже вместо sshuser укажите реальное имя пользователя, если оно отличается. Вместо mycluster укажите реальное имя кластера. Рабочей должна быть папка, в которой находится файл.

Используйте scp для копирования файлов в кластер HDInsight. Измените и введите команду :

scp pigudf.py sshuser@mycluster-ssh.azurehdinsight.net: 

ssh sshuser@mycluster-ssh.azurehdinsight.net 

hdfs dfs -put pigudf.py /pigudf.py 

Использование Pig UDF (оболочка)

1. Чтобы подключиться к Pig, в открытом сеансе SSH введите следующую команду:

Register wasbs:///pigudf.py using jython as myfuncs; LOGS = LOAD 'wasbs:///example/data/sample.log' as (LINE:chararray); LOG = FILTER LOGS by LINE is not null; DETAILS = foreach LOG generate myfuncs.create_structure(LINE); DUMP DETAILS; 

((2012-02-03,20:11:56,SampleClass5,[TRACE],verbose detail for id 990982084)) ((2012-02-03,20:11:56,SampleClass7,[TRACE],verbose detail for id 1560323914)) ((2012-02-03,20:11:56,SampleClass8,[DEBUG],detail for id 2083681507)) ((2012-02-03,20:11:56,SampleClass3,[TRACE],verbose detail for id 1718828806)) ((2012-02-03,20:11:56,SampleClass3,[INFO],everything normal for id 530537821)) 

nano pigudf.py 

#from pig_util import outputSchema 

Register 'pigudf.py' using streaming_python as myfuncs; LOGS = LOAD 'wasbs:///example/data/sample.log' as (LINE:chararray); LOG = FILTER LOGS by LINE is not null; DETAILS = foreach LOG generate myfuncs.create_structure(LINE); DUMP DETAILS; 

Отправка файла (PowerShell)

PowerShell также можно использовать для удаленного запуска запросов на использование Hive. Рабочей должна быть папка, в которой находится pigudf.py . Чтобы выполнить запрос Hive, использующий скрипт pigudf.py , примените следующий скрипт PowerShell:

# Login to your Azure subscription # Is there an active Azure subscription? $sub = Get-AzSubscription -ErrorAction SilentlyContinue if(-not($sub)) < Connect-AzAccount ># If you have multiple subscriptions, set the one to use # Select-AzSubscription -SubscriptionId "" # Revise file path as needed $pathToJythonFile = ".\pigudf.py" # Get cluster info $clusterName = Read-Host -Prompt "Enter the HDInsight cluster name" $clusterInfo = Get-AzHDInsightCluster -ClusterName $clusterName $resourceGroup = $clusterInfo.ResourceGroup $storageAccountName=$clusterInfo.DefaultStorageAccount.split('.')[0] $container=$clusterInfo.DefaultStorageContainer $storageAccountKey=(Get-AzStorageAccountKey ` -ResourceGroupName $resourceGroup ` -Name $storageAccountName)[0].Value # Create an Azure Storage context $context = New-AzStorageContext ` -StorageAccountName $storageAccountName ` -StorageAccountKey $storageAccountKey # Upload local files to an Azure Storage blob Set-AzStorageBlobContent ` -File $pathToJythonFile ` -Blob "pigudf.py" ` -Container $container ` -Context $context 

Использование Pig UDF (PowerShell)

При удаленной отправке задания с помощью PowerShell нельзя использовать CPython в качестве интерпретатора.

PowerShell также можно использовать для запуска заданий Pig Latin. Чтобы выполнить задание Pig Latin, использующее скрипт pigudf.py , примените следующий скрипт PowerShell:

# Script should stop on failures $ErrorActionPreference = "Stop" # Login to your Azure subscription # Is there an active Azure subscription? $sub = Get-AzSubscription -ErrorAction SilentlyContinue if(-not($sub)) < Connect-AzAccount ># Get cluster info $clusterName = Read-Host -Prompt "Enter the HDInsight cluster name" $creds=Get-Credential -UserName "admin" -Message "Enter the login for the cluster" $PigQuery = "Register wasbs:///pigudf.py using jython as myfuncs;" + "LOGS = LOAD 'wasbs:///example/data/sample.log' as (LINE:chararray);" + "LOG = FILTER LOGS by LINE is not null;" + "DETAILS = foreach LOG generate myfuncs.create_structure(LINE);" + "DUMP DETAILS;" # Create Pig job object $jobDefinition = New-AzHDInsightPigJobDefinition -Query $PigQuery # For status bar updates $activity="Pig job" # Progress bar (optional) Write-Progress -Activity $activity -Status "Starting job. " # Start defined Azure HDInsight job on specified cluster. $job = Start-AzHDInsightJob ` -ClusterName $clusterName ` -JobDefinition $jobDefinition ` -HttpCredential $creds # Progress bar (optional) Write-Progress -Activity $activity -Status "Waiting for the Pig job to complete. " # Wait for completion or failure of specified job Wait-AzHDInsightJob ` -Job $job.JobId ` -ClusterName $clusterName ` -HttpCredential $creds # Uncomment the following to see stderr output # Progress bar (optional) Write-Progress -Activity $activity "Retrieving output. " # Gets the log output Get-AzHDInsightJobOutput ` -Clustername $clusterName ` -JobId $job.JobId ` -HttpCredential $creds 

Результат выполнения задания Pig должен выглядеть аналогично следующим данным:

((2012-02-03,20:11:56,SampleClass5,[TRACE],verbose detail for id 990982084)) ((2012-02-03,20:11:56,SampleClass7,[TRACE],verbose detail for id 1560323914)) ((2012-02-03,20:11:56,SampleClass8,[DEBUG],detail for id 2083681507)) ((2012-02-03,20:11:56,SampleClass3,[TRACE],verbose detail for id 1718828806)) ((2012-02-03,20:11:56,SampleClass3,[INFO],everything normal for id 530537821)) 

Устранение неполадок

Ошибки при выполнении заданий

При выполнении задания hive может возникнуть ошибка, аналогичная приведенной ниже:

Caused by: org.apache.hadoop.hive.ql.metadata.HiveException: [Error 20001]: An error occurred while reading or writing to your custom script. It may have crashed with an error. 

Эта проблема может быть вызвана символами окончания строк в файле Python. Многие редакторы Windows по умолчанию используют символы CRLF, но в приложениях Linux обычно ожидается использование символа LF.

Вы можете использовать следующие команды PowerShell для удаления символов CR перед передачей файла в HDInsight:

Write-Progress -Activity $activity -Status "Waiting for the Pig job to complete. " # Wait for completion or failure of specified job 

Сценарии PowerShell

Оба примера скриптов PowerShell, используемых для запуска примеров, содержат закомментированную строку, которая отображает вывод ошибок для задания. Если вы не видите ожидаемых результатов задания, раскомментируйте следующую строку и посмотрите, нет ли в описании ошибки информации о проблеме.

$activity="Pig job" # Progress bar (optional) Write-Progress -Activity $activity -Status "Starting job. " 

Сведения об ошибках (STDERR) и результат выполнения задания (STDOUT) также записываются в хранилище HDInsight.

Дальнейшие действия

Если вам нужно загрузить модули Python, которые не поставляются по умолчанию, см. статью How to deploy a Python module to Windows Azure HDInsight (Как развернуть модуль Python в Windows Azure HDInsight).

Сведения о других способах использования Pig и Hive и дополнительную информацию об использовании MapReduce см. в следующих документах:

• Использование Hive и HiveQL с Hadoop в HDInsight для анализа примера файла Apache log4j
• Использование MapReduce в Hadoop в HDInsight

Big Data от А до Я. Часть 2: Hadoop

Привет, Хабр! В предыдущей статье мы рассмотрели парадигму параллельных вычислений MapReduce. В этой статье мы перейдём от теории к практике и рассмотрим Hadoop – мощный инструментарий для работы с большими данными от Apache foundation.

В статье описано, какие инструменты и средства включает в себя Hadoop, каким образом установить Hadoop у себя, приведены инструкции и примеры разработки MapReduce-программ под Hadoop.

Общая информация о Hadoop

Как известно парадигму MapReduce предложила компания Google в 2004 году в своей статье MapReduce: Simplified Data Processing on Large Clusters. Поскольку предложенная статья содержала описание парадигмы, но реализация отсутствовала – несколько программистов из Yahoo предложили свою реализацию в рамках работ над web-краулером nutch. Более подробно историю Hadoop можно почитать в статье The history of Hadoop: From 4 nodes to the future of data

Изначально Hadoop был, в первую очередь, инструментом для хранения данных и запуска MapReduce-задач, сейчас же Hadoop представляет собой большой стек технологий, так или иначе связанных с обработкой больших данных (не только при помощи MapReduce).

Основными (core) компонентами Hadoop являются:

Hadoop Distributed File System (HDFS) – распределённая файловая система, позволяющая хранить информацию практически неограниченного объёма.

Hive – инструмент для SQL-like запросов над большими данными (превращает SQL-запросы в серию MapReduce–задач);

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

Установка Hadoop на кластер при помощи Cloudera Manager

Раньше установка Hadoop представляла собой достаточно тяжёлое занятие – нужно было по отдельности конфигурировать каждую машину в кластере, следить за тем, что ничего не забыто, аккуратно настраивать мониторинги. С ростом популярности Hadoop появились компании (такие как Cloudera, Hortonworks, MapR), которые предоставляют собственные сборки Hadoop и мощные средства для управления Hadoop-кластером. В нашем цикле материалов мы будем пользоваться сборкой Hadoop от компании Cloudera.

Для того чтобы установить Hadoop на свой кластер, нужно проделать несколько простых шагов:

1. Скачать Cloudera Manager Express на одну из машин своего кластера отсюда;
2. Присвоить права на выполнение и запустить;
3. Следовать инструкциям установки.

После установки вы получите консоль управления кластером, где можно смотреть установленные сервисы, добавлять/удалять сервисы, следить за состоянием кластера, редактировать конфигурацию кластера:

Более подробно с процессом установки Hadoop на кластер при помощи cloudera manager можно ознакомиться по ссылке в разделе Quick Start.

Если же Hadoop планируется использовать для «попробовать» – можно не заморачиваться с приобретением дорогого железа и настройкой Hadoop на нём, а просто скачать преднастроенную виртуальную машину по ссылке и пользоваться настроенным hadoop’ом.

Запуск MapReduce программ на Hadoop

Теперь покажем как запустить MapReduce-задачу на Hadoop. В качестве задачи воспользуемся классическим примером WordCount, который был разобран в предыдущей статье цикла. Для того, чтобы экспериментировать на реальных данных, я подготовил архив из случайных новостей с сайта lenta.ru. Скачать архив можно по ссылке.

Напомню формулировку задачи: имеется набор документов. Необходимо для каждого слова, встречающегося в наборе документов, посчитать, сколько раз встречается слово в наборе.

def map(doc): for word in doc.split(): yield word, 1

def reduce(word, values): yield word, sum(values) 

Способ №1. Hadoop Streaming

Самый простой способ запустить MapReduce-программу на Hadoop – воспользоваться streaming-интерфейсом Hadoop. Streaming-интерфейс предполагает, что map и reduce реализованы в виде программ, которые принимают данные с stdin и выдают результат на stdout.

Программа, которая исполняет функцию map называется mapper. Программа, которая выполняет reduce, называется, соответственно, reducer.

Streaming интерфейс предполагает по умолчанию, что одна входящая строка в mapper или reducer соответствует одной входящей записи для map.

Вывод mapper’a попадает на вход reducer’у в виде пар (ключ, значение), при этом все пары соответствующие одному ключу:

• Гарантированно будут обработаны одним запуском reducer’a;
• Будут поданы на вход подряд (то есть если один reducer обрабатывает несколько разных ключей – вход будет сгруппирован по ключу).

#mapper.py import sys def do_map(doc): for word in doc.split(): yield word.lower(), 1 for line in sys.stdin: for key, value in do_map(line): print(key + "\t" + str(value)) 

#reducer.py import sys def do_reduce(word, values): return word, sum(values) prev_key = None values = [] for line in sys.stdin: key, value = line.split("\t") if key != prev_key and prev_key is not None: result_key, result_value = do_reduce(prev_key, values) print(result_key + "\t" + str(result_value)) values = [] prev_key = key values.append(int(value)) if prev_key is not None: result_key, result_value = do_reduce(prev_key, values) print(result_key + "\t" + str(result_value)) 

Данные, которые будет обрабатывать Hadoop должны храниться на HDFS. Загрузим наши статьи и положим на HDFS. Для этого нужно воспользоваться командой hadoop fs:

wget https://www.dropbox.com/s/opp5psid1x3jt41/lenta_articles.tar.gz tar xzvf lenta_articles.tar.gz hadoop fs -put lenta_articles

Утилита hadoop fs поддерживает большое количество методов для манипуляций с файловой системой, многие из которых один в один повторяют стандартные утилиты linux. Подробнее с её возможностями можно ознакомиться по ссылке.

Теперь запустим streaming-задачу:

yarn jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar\ -input lenta_articles\ -output lenta_wordcount\ -file mapper.py\ -file reducer.py\ -mapper "python mapper.py"\ -reducer "python reducer.py"

Утилита yarn служит для запуска и управления различными приложениями (в том числе map-reduce based) на кластере. Hadoop-streaming.jar – это как раз один из примеров такого yarn-приложения.

Дальше идут параметры запуска:

• input – папка с исходными данными на hdfs;
• output – папка на hdfs, куда нужно положить результат;
• file – файлы, которые нужны в процессе работы map-reduce задачи;
• mapper – консольная команда, которая будет использоваться для map-стадии;
• reduce – консольная команда которая будет использоваться для reduce-стадии.

В интерфейсе доступном по этому URL можно узнать более детальный статус выполнения задачи, посмотреть логи каждого маппера и редьюсера (что очень полезно в случае упавших задач).

Результат работы после успешного выполнения складывается на HDFS в папку, которую мы указали в поле output. Просмотреть её содержание можно при помощи команды «hadoop fs -ls lenta_wordcount».

Сам результат можно получить следующим образом:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5 с 41 что 43 на 82 и 111 в 194

Команда «hadoop fs -text» выдаёт содержимое папки в текстовом виде. Я отсортировал результат по количеству вхождений слов. Как и ожидалось, самые частые слова в языке – предлоги.

Способ №2

Сам по себе hadoop написан на java, и нативный интерфейс у hadoop-a тоже java-based. Покажем, как выглядит нативное java-приложение для wordcount:

import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordCount < public static class TokenizerMapper extends Mapper < private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException < StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) < word.set(itr.nextToken()); context.write(word, one); >> > public static class IntSumReducer extends Reducer  < private IntWritable result = new IntWritable(); public void reduce(Text key, Iterablevalues, Context context ) throws IOException, InterruptedException < int sum = 0; for (IntWritable val : values) < sum += val.get(); >result.set(sum); context.write(key, result); > > public static void main(String[] args) throws Exception < Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_articles")); FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_wordcount")); System.exit(job.waitForCompletion(true) ? 0 : 1); >>

Этот класс делает абсолютно то же самое, что наш пример на Python. Мы создаём классы TokenizerMapper и IntSumReducer, наследуя их от классов Mapper и Reducer соответсвенно. Классы, передаваемые в качестве параметров шаблона, указывают типы входных и выходных значений. Нативный API подразумевает, что функции map на вход подаётся пара ключ-значение. Поскольку в нашем случае ключ пустой – в качестве типа ключа мы определяем просто Object.

В методе Main мы заводим mapreduce-задачу и определяем её параметры – имя, mapper и reducer, путь в HDFS, где находятся входные данные и куда положить результат.

Для компиляции нам потребуются hadoop-овские библиотеки. Я использую для сборки Maven, для которого у cloudera есть репозиторий. Инструкции по его настройке можно найти по ссылке. В итоге файл pom.xmp (который используется maven’ом для описания сборки проекта) у меня получился следующий):

  4.0.0  cloudera https://repository.cloudera.com/artifactory/cloudera-repos/    org.apache.hadoop hadoop-common 2.6.0-cdh5.4.2  org.apache.hadoop hadoop-auth 2.6.0-cdh5.4.2  org.apache.hadoop hadoop-hdfs 2.6.0-cdh5.4.2  org.apache.hadoop hadoop-mapreduce-client-app 2.6.0-cdh5.4.2   org.dca.examples wordcount 1.0-SNAPSHOT 

Соберём проект в jar-пакет:

mvn clean package

После сборки проекта в jar-файл запуск происходит похожим образом, как и в случае streaming-интерфейса:

yarn jar wordcount-1.0-SNAPSHOT.jar WordCount

Дожидаемся выполнения и проверяем результат:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5 с 41 что 43 на 82 и 111 в 194

Как нетрудно догадаться, результат выполнения нашего нативного приложения совпадает с результатом streaming-приложения, которое мы запустили предыдущим способом.

Резюме

В статье мы рассмотрели Hadoop – программный стек для работы с большими данными, описали процесс установки Hadoop на примере дистрибутива cloudera, показали, как писать mapreduce-программы, используя streaming-интерфейс и нативный API Hadoop’a.

В следующих статьях цикла мы рассмотрим более детально архитектуру отдельных компонент Hadoop и Hadoop-related ПО, покажем более сложные варианты MapReduce-программ, разберём способы упрощения работы с MapReduce, а также ограничения MapReduce и как эти ограничения обходить.

Спасибо за внимание, готовы ответить на ваши вопросы.

Ссылки на другие статьи цикла: