Big Data

В дистрибутиве от Cloudera напрямую работать с supervisor не получится. Дело в том, что саму службу они скрыли и она не доступна по стандартной команде "service supervisord" а файли настроек не находятся на своем привычном месте "/etc/supervisor/supervisord.conf". Так что же делать, если у Вас возникла например необходимость настроить стабильную работу  REST API для работы c Spark-ом на основе Livy Spark Server  ? Проблема усложняется тем, что на данный момент дистрибутив Cloudera официально не поддерживает эту службу и потому нужно самому собрать и запустить Livy сервер из исходников. После установки самого Livy сервера возникает необходимость  повысить отказоустойчивость путём мониторинга и автоматического перезапуска службы в случае падения. Настройки supervisor-a в случае собранного кластера на основе дистрибутива от Cloudera находится по этому адресу: "/var/run/cloudera-scm-agent/supervisor/include/" Создадим в этой директории файл "li...

В Hue 3.8 появилясь возможность использовать интерактивный Web-based блокнот для анализа данных " Notebook Application for Spark & SQL ". Вдохновление ребята взяли с аналогичной разработки  IPython Notebook. Идея в том, чтобы избавить аналитиков от изнурительной работы над компиляцией Jar файлов и их запуска с консоли. Это очень удобно, когда нужно сделать быструю проверку данных, проверку теории и т.д.. Проблема в том, что по умолчанию для запуска этого блокнота должен быть настроен Livy Spark Server. Об этом нам говорит следующее сообщение: "The app won't work without a running Livy Spark Server" В настройках hue.ini устанавливаем следующее: [desktop] app_blacklist= [notebook] show_notebooks=true [beeswax] use_new_editor=true Скачиваем и устанавливаем Livy server https://github.com/cloudera/livy  cd /usr/local/src git clone git@github.com:cloudera/livy.git cd livy-master export JAVA_HOME=/usr/java/jdk1.7.0_67-cloudera export PATH=$JAVA_HO...

Загружаем spark-shell и указываем путь к jdbc драйверу для работы с Impala : spark-shell --master "local[2]" --driver-class-path /opt/cloudera/parcels/CDH/jars/ImpalaJDBC41.jar После запуска spark-shell нужно импортировать все зависимости и октрыть соединение с помощью DriverManager. Далее приведены примеры чтения с таблицы и добавления данных: import java.util.Properties import org.apache.spark._ import org.apache.spark.sql.SQLContext import java.sql.Connection import java.sql.DriverManager Class.forName("com.cloudera.impala.jdbc41.Driver") var conn: java.sql.Connection = null conn = DriverManager.getConnection("jdbc:impala://127.0.0.1:21050/default;auth=noSasl", "", "") val statement = conn.createStatement(); val result = statement.executeQuery("SELECT * FROM users limit 10") result.next() result.getString("user_id")val sql_insert = "INSERT INTO users VALUES('user_id','email','gend...

import org.apache.hadoop.fs._ import org.apache.spark.deploy.SparkHadoopUtil import java.net.URI val hdfs_conf = SparkHadoopUtil.get.newConfiguration(sc.getConf) val hdfs = FileSystem.get(hdfs_conf) // source data in HDFS val sourcePath = new Path("/data/*/*") hdfs.globStatus( sourcePath ).foreach{ fileStatus => val filePathName = fileStatus.getPath().toString() val fileName = fileStatus.getPath().getName() println("fileName: " + fileName + "; FilePath: " + filePathName) val fileRDD = sc.textFile(filePathName) val counts = fileRDD.flatMap(line => line.split("\n")) //val df = sqlContext.read.json( filePathName) //df.show() counts.foreachPartition(partF => { partF.foreach( item_row => { println(item_row) }); }); }

Эта ошибка возникает в ситуациях, когда spark-streaming задача восстанавливается с последнего checkpoint-a. 1-е нужно у убрать с кода обращение к значению ".value". Тоесть читать значение акумулятора не нужно через значение ".value". Нужно обращаться просто по имени акумулятора. Если вы используете checkpoint-ы, при восстановлении работы драйвера, акумулятор не реинициализирован, потому вы можете увидеть следующую ошибку: ERROR scheduler.DAGScheduler: Failed to update accumulators for ResultTask(5, 5) java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Long Ошибка эта также говорит нам о том, что акумулятор не реиницилизирован при восстановлении с checkpoint-a. Для этого нужно добавить проверку на инициализацию акумулятора: object Accum_inst { @volatile private var instance: Accumulator[Long] = null def getInstance(sc: SparkContext, AccumName: String): Accumulator[Long] = { if (instance == null) { synchronized { ...

SparkOnHbase https://github.com/cloudera-labs/SparkOnHBase Скачиваем jar архив spark-hbase-0.0.2-clabs.jar для использования возможностей SparkOnHbase Запускаем spark-shell: spark-shell --master "local[2]" --driver-class-path /home/hdfs/spark-hbase-0.0.2-clabs.jar,/opt/cloudera/parcels/CDH/lib/hbase/lib/htrace-core-3.1.0-incubating.jar import com.cloudera.spark.hbase._ import org.apache.hadoop.hbase.spark.HBaseContext import org.apache.hadoop.hbase.{TableName, HBaseConfiguration} import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.client.Scan import org.apache.spark.sql.SQLContext import org.apache.spark.SparkContext import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.fs.Path import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.client.Get import org.apache.hadoop.hbase.client.Result import org.apache.spark.SparkConf val rdd = sc.parallelize(Array( (Bytes.toBytes("user_id:10")), (By...

   Одним из эффективных способов существенного ускорения вашего Spark Streaming приложения, являвляется применения кеширования RDD а также конфигурация способа кеширования ваших данных.    При использовании rdd.cache первая операция по rdd будет занимать тоже время что и без применения кешированя, но все последующие операции будут занимать значительно меньше времени. Это может быть удивительно, ведь Spark стараеться делать все операции с данными в памяти. Это сложно понять, но приминение rdd.cache имеет значимый эффект..    Тесты показывают, что если DStream или RDD используется несколько раз, то их кеширование значительно увеличивает скорость их обработки. Возможно это будет являться тем изменением, что принесёт вам наибольшее улучшнеием производительности. Пример применения rdd.cache: dstream.foreachRDD{rdd => rdd.cache() // cache the RDD before iterating! rdd.foreach{ key => rdd.filter(elem=> key(elem) == key).saveAsFooBar(....

Всего несколько параметров позволять Вам существенно увеличить производительность Spark Streaming приложений, которые читают топики с Kafka и записывают их в хранилище(например в HDFS). Первым, что стоит оптимизировать, это получателя сообщений с Kafka. Таким получателем выступает Ваше Spark Streaming приложение, которое читает сообщения с топика Kafka порциями(блоками, партициями). Размер этих блоков нужно настраивать относительно предположительной скорости сохранения данных в хранилище. По умолчанию размер этих блоков не установлен и его нужно определить исходя из скорости обработки сообщений Вашим получателем. Вот формула по которой можно понять, сколько же сообщений за раз получит Ваш воркер: partitionSize = (1000 / blockInterval) * maxRate После нескольких запусков Вы уже приблизительно понимаете сколько записей в секунду обрабатывает Ваш получатель. К примеру мы знаем, что воркер справляется приблизительно с 1000 записей в секунду. Возьмём половину от этого и установим ...

Запускаем spark-shell и выполняем запись в локальный файл на сервере: import java.io.File import com.google.common.io.Files import java.nio.charset.Charset val outputPath = "/home/hdfs/src/HBase/last_key.txt" val outputFile = new File(outputPath) println("Appending to " + outputFile.getAbsolutePath) Files.append("Test message for save into hdfs!!", outputFile, Charset.defaultCharset())

Ниже приведён код инициализации соединения и выбора данных с базы данных через jdbc DriveManager: import org.apache.spark._ import org.apache.spark.sql.SQLContext import java.sql.{Connection, DriverManager, ResultSet} val conn_str = "jdbc:vertica://127.0.0.1:5433/testdbname?user=testuser&password=testpass" var conn: java.sql.Connection = null conn = DriverManager.getConnection(conn_str); val statement = conn.createStatement() val resultSet = statement.executeQuery("SELECT * FROM public_chema.users WHERE id=777") resultSet.next() println("data from id: " + resultSet.getString(1)) println("data from email: " + resultSet.getString(2)) //make update or insert val sql_update = "UPDATE public_chema.test_table SET email = '" + resultSet.getString(2) + "' WHERE id=" + resultSet.getString(1) //val sql_update = "INSERT INTO public_chema.test_table (email, id) VALUES ('" + resultSet.getString(2) + "...

Проблема возникает когда вы пытаетесь запустить Apache Spark приложение в режиме yarn cluster. Пример запуска: spark-submit --master yarn --deploy-mode cluster ... В настройках yarn-a установлено кол. попыток запуска задачи. Когда первая попытка запустить приложение терпит неудачу, то Yarn пытается запустить приложение повторно, но терпит неудачу из-за ошибки "Target log file already exists (/user/spark/applicationHistory/application_ .". Эта ошибка означает, что включена переменная spark.eventLog.enabled, которая говорит о том, что в директории, которая указана в переменной spark.eventLog.dir будут храниться логи событий при запуске приложений на класетере YARN. Решается эта проблема довольно интерестно. Дело в том, что в Spark-e есть незадокументированная переменная spark.eventLog.overwrite, которая и отвечает за возможность перезаписи файла логов событий и ошибка перестаёт возникать. Но эту настройку не так-то просто и установить. Вариант 1: В коде приложения ...

При попытке записи данных в HBase возникла проблема запуска кода через spark-submit. При этом этот же код правильно отрабатывал локально через spark-shell. Ошибка следующая: "java.io.IOException: java.lang.reflect.InvocationTargetException... Caused by: java.lang.NoClassDefFoundError: org/apache/htrace/Trace" Проблема в том, что Spark не включает в себя сторонние зависимости. Тоесть по умолчанию он не знает о местоположении jars архивов для драйвера HBase. Интерестно то что Spark включает в себя все необходимые зависимости для работы с Hive и для работы с этой БД. ничего подключать дополнительно не нужно. Для решения этой несправедливости необходимо добавить в classpath путь к jar-архиву драйвера к HBase. Вот как это должно выглядеть: spark-submit --master yarn-cluster --driver-class-path /etc/hbase/conf:/opt/cloudera/parcels/CDH/lib/hbase/lib/htrace-core-3.1.0-incubating.jar ... Есть также возможность добавить путь к Jar архиву в конфиг Spark-a. Для этого в файл  /e...

Этот пример показывает как можно быстро сформировать строку для INSERT запроса в базу данных: import scala.collection.mutable.ArrayBuffer var sql_insert_arr = ArrayBuffer[String]() var sql_insert_str = "INSERT INTO test_table VALUES " sql_insert_arr += "('user_id_1', 'user_email_1', 'user_age_1')" sql_insert_arr += "('user_id_2', 'user_email_2', 'user_age_2')" sql_insert_str = sql_insert_str + sql_insert_arr.mkString(",")

Загружаем spark-shell: spark-shell --master "local[2]" После запуска spark-shell нужно импортировать все зависимости и октрыть соединение с помощью DriverManager. Далее приведены примеры чтения с таблицы и добавления данных: import java.util.Properties import org.apache.spark._ import org.apache.spark.sql.SQLContext import java.sql.Connection import java.sql.DriverManager Class.forName("org.apache.hive.jdbc.HiveDriver") var conn: java.sql.Connection = null conn = DriverManager.getConnection("jdbc:hive2://127.0.0.1:10000/default", "", "") val statement = conn.createStatement() val result = statement.executeQuery("SELECT * FROM users_registered") result.next() result.getString("email") val sql_insert = "INSERT INTO users_registered VALUES('user_id','email','gender')" statement.executeUpdate(sql_insert)

Ниже приведён код инициализации соединения и выбора данных с базы данных. После инициализации мы получаем dataFrame: import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ val sqlContext = new org.apache.spark.sql.SQLContext(sc) import sqlContext.implicits._ val jdbcDF = sqlContext.load("jdbc", Map("url" -> "jdbc:postgresql://127.0.0.1:5432/yourdb?user=dbuser&password=userpassword", "dbtable" -> "public_chema.table_name")) val data_by_channel_id = jdbcDF.filter(jdbcDF("channel_id") === 265) data_by_channel_id.show() Также возможен альтернативный способ работы с базой через DriveManager: import java.util.Properties import org.apache.spark._ import org.apache.spark.sql.SQLContext import java.sql.Connection import java.sql.DriverManager val url = "jdbc:postgresql://127.0.0.1:5432/dbname" var conn: java.sql.Connection = null val properties = new Properties() properties.setProperty(...

Краткий обзор разницы между тем как 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, интерактивная сессия с нескольким...

В этот статье приводится крактий обзор того, как Spark запускается на кластере, для того чтобы проще понять компоненты которые системы. Компоненты: Spark приложение запускается как массив независимых процессов на кластере, координацией которых занимается объект SparkContext в вашей программе(называется она driver program). В режиме кластера, SparkContext может работать в паре с одним из нескольких менеджеров кластера (собственный менеджер менеджер кластера, Мesos или YARN ), который выделяет ресурсы для приложений. Единожды присоединившись, Spark получает исполнителей (executors) на ноде в кластере, которые обрабатывают вычисления и сохраняют данные для вашего приложения. В итоге, SparkContext отсылает задачи (tasks) к исполнителям для обработки. Нужно отметить несколько полезных особенностей в этой архитектуре: Каждое приложение получает свой собственный процес-исполнитель, который остается запущеным на продолжении всего времени работы приложения и запускает таски в многоп...

Загрузка данных в 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, на локальную машину или в любую другую поддерживаемую файловую систему — полный список можно посмотреть в докумен...

Apache Spark SQL - позволяет относительно лего смешать SQL запросы с Spark программой Поддерживает разные способы доступа включая Hive, Avro, Parquet, ORC, JSON, and JDBC or ODBC стандарты Сильным приемуществом  Spark SQL является его глубокая интеграция с Spark, позволяющий использовать другие библиотеки( MLlib ,  Spark streaming , GraphX и т.д.) очень просто. Пример кода: df_ads = sqlContext.read.format("com.databricks.spark.csv").options(delimiter="\t").options(header="true").load("/Volumes/work/data/kaggle/avitocontextadd/AdsInfo.tsv") df_ads.registerTempTable("ads") df_test = sqlContext.read.format("com.databricks.spark.csv").options(delimiter="\t").options(header="true").load("/Volumes/work/data/kaggle/avitocontextadd/testSearchStream.tsv") #Filter only contextual data from training dataset trainingdata = df_test.filter(df_test["ObjectType"] == 3) trainingdata.re...

Apache Spark - фреймворк (набор свыше 80-ти операторов(функций) для работы с данными (filter, distinct, union, intersection, reduce и.т.д.)), который разрабатывался для распределенной обработки больших данных. Основным преимуществом данного инструмента, является то, что данные можно кэшировать в оперативной памяти, что позволяет существенно ускорять вычисления в случае итеративных алгоритмов, какими и являются большинство алгоритмов машинного обучения. Суть его в том, что это переосмысленный Map&Reduce 2.0(от Hadoop) и работает он в 10x-100x раз быстрее в зависимости от работы в памяти или на диске. Маленький троллинг Hadoop-a от команды Spark-a: Нативно Spark поддерживает Scala, Python, Java, R. Очень удобно работать непосредственно в IPython Notebook, выгружая небольшую часть данных из кластера и обрабатывая, например, пакетом Pandas — получается довольно удобная связка. Итак, начнем с того, что основным понятием в Spark'е является RDD (Resilient Distributed Dataset), ...