а кто-нибудь пробовал Spark streaming юзать с интервалом в пару сотень миллисекунд? Это вообще жизнеспособно?

зависит от того что ты кормишь ей

и как кормишь*

ну если это structured streaming из кафки в jdbc

с парой сотней колонок

просто переложить надо?

просто переложить надо?

не. там 1 к 1 трансформации + зааплаить модель, что по сути тоже 1 в 1 трансформация. То есть кафка стрим -> пайплайн линейный -> jdbc

кафка коннектор заюзать и не тащить спарк

трансформации же в итоге в кафка стримс, коннектором сложить по jdbc

спарк здесь принципиально нужен

так как весь ETL построен на DataFrame'ах уже

ну если это structured streaming из кафки в jdbc

а что за база? и как она живет?

а что за база? и как она живет?

проблема в том, что может быть что угодно. Хоть редшифт, хоть постгресс. Какой драйвер подсунут - то и будет. То есть я понимаю что итоговый сроупут от этого очень зависит, но я пока пытаюсь понять какой теоретический сроупут вообще можно из этого выжать. Задача предоставить такой near real time, что прям вот near near. При этом 100 миллисекунд латенси - вроде как допустима, по крайней мере на это число ориентир

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

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

Понял. Спасибо

и снова по apache phoenix — кто-то юзал его batch upsert ?

насущный вопрос - есть long-running application в спарке запущенном в cluster-mode на Amazon EMR. Как бы его получше мониторить?

т.е. рестарты вроде кластер сам обеспечит

а вот если после рестартов, приложение упало перманентно. кто-то занимался таким оповещением?

стриминг ?

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

насущный вопрос - есть long-running application в спарке запущенном в cluster-mode на Amazon EMR. Как бы его получше мониторить?

у приложения есть свой ID (его можно получить через yarn application -list) Дальше уже вариантов много - можно скриптами проверять, можно через API YARN всю информацию о приложении получать

если амбари есть, то там есть свой мониторинг

но у вас EMR

ну есть MR каунтеры, их можно мониторить в тч их изменение - будет понятно висит job-а или нет

удивлен что надо самому что-то придумывать (

Беда, но иногда в разработке такое случается, что поделать ?

ну есть MR каунтеры, их можно мониторить в тч их изменение - будет понятно висит job-а или нет

я бы не подвязывался

мне кажется проще делать метрики прям в джобе)

тут вопрос куда писать эти метрики, mr job-то распределенный. counter-ы это по сути и есть агрегация данных с каждого маппера-редусера

ну опентсдб

в org.elasticsearch.spark.sql параметр spark.es.http.timeout работает у кого нибудь?

что 5секунд его ставь, что 10 минут, таймаут происходит ровно через 2 минуты

Всем привет, есть вопросик, возможно нубский, насчёт обработки данных спарком. Есть метрики записанные в csv формата (metric_id, object_id, timestamp, value). Сочетание (metric_id, object_id, timestamp) можно считать уникальным. Надо преобразовать csv таким образом чтобы в конечном итоге получился dataframe - (object_id, timestamp, <metric_0>, <metric_1>, ..., <metric_N>), где названия колонок metric_* известны заранее и совпадают со значениями из metric_id, а значения в этих колонках берутся из value. Хотелось бы собирать такой датафрейм в один проход по csv, например, создавать RDD такой же структуры как csv и потом делать filter -> map -> groupBy(object_id, timestamp) -> sum(*metrics). Это работает если значения численные - во время map пишу ноль во все метрики кроме данной и потом каждое значение складывается с нолями в groupBy -> sum. Но это не поможет с текстовыми значениями и вообще выглядит костыльно. Может есть более правильный способ решения такой задачи?

https://databricks.com/blog/2016/02/09/reshaping-data-with-pivot-in-apache-spark.html - не оно?

https://databricks.com/blog/2016/02/09/reshaping-data-with-pivot-in-apache-spark.html - не оно?

Близко, да, но там всё равно в итоге надо агрегацию делать по значениям. Сейчас пробую groupByKey() использовать для rdd вида ((object_id, timestamp), (metric_id, value)). Там можно любую функцию агрегации взять, в моём случае достаточно записывать в один словарь. Надеюсь, оно так сработает.

https://www.humblebundle.com/books/data-science-books

надо брать

найс

Всем привет, есть вопросик, возможно нубский, насчёт обработки данных спарком. Есть метрики записанные в csv формата (metric_id, object_id, timestamp, value). Сочетание (metric_id, object_id, timestamp) можно считать уникальным. Надо преобразовать csv таким образом чтобы в конечном итоге получился dataframe - (object_id, timestamp, <metric_0>, <metric_1>, ..., <metric_N>), где названия колонок metric_* известны заранее и совпадают со значениями из metric_id, а значения в этих колонках берутся из value. Хотелось бы собирать такой датафрейм в один проход по csv, например, создавать RDD такой же структуры как csv и потом делать filter -> map -> groupBy(object_id, timestamp) -> sum(*metrics). Это работает если значения численные - во время map пишу ноль во все метрики кроме данной и потом каждое значение складывается с нолями в groupBy -> sum. Но это не поможет с текстовыми значениями и вообще выглядит костыльно. Может есть более правильный способ решения такой задачи?

Если известны заранее, то можно разделить датафрейм на несколько датафреймов с разными метриками и поджойнить их по ключу object_id. В целом это обработка мультилайна получается

если я все правильно понял

Первый dataframe будет val objects = df.select('object_id, 'timestamp).distinct

дальше будет несколько df с метриками типа val metric0 = df.filter('metric_id === x).select('object_id, 'value.alias("metric_0"))

потом джойн val joined = objects.join(metric0, Seq("object_id"))

так на выходе получишь таблицу, в которой уникальный object_id будет только в одной строчке и все твои метрики будут в ней же

дальше будет несколько df с метриками типа val metric0 = df.filter('metric_id === x).select('object_id, 'value.alias("metric_0"))

Тут нужно ещё timestamp в селект добавить, верно?

там еще value надо добавить

А, ну да

а во втором тебе timestamp не нужен

тк он в objects будет

Ага, понял

ну это если у тебя timestamp один. Если он не один, то надо понимать, какой ты хочешь врезультирущей таблице

ну это если у тебя timestamp один. Если он не один, то надо понимать, какой ты хочешь врезультирущей таблице

Таймстамп один на несколько метрик и много для одного объекта. Повторные метрики с тем же таймстампом можно отбрасывать.

тогда его можно в objects

а в остальных df отбрасывать

Вообще вроде разобрался, спасибо. Значит, делать много датафреймов и джойнить их это норм? С точки зрения производительности

они большие?

Ну, да, там на самом деле не один csv а много просто все имеют одинаковую структуру

а в количестве событий на джобу есть оценка?

Нет, а как можно посчитать?