а зукипер какой (версия)?

те я могу вызывать validatePath в своем коде и он отрабатывает

еще подумал: мб там у тебя эфимерная нода, не?

а зукипер какой (версия)?

3.4.5

еще подумал: мб там у тебя эфимерная нода, не?

не, нода обычная

3.4.5

правильно либу подключил?

https://curator.apache.org/zk-compatibility.html

правильно либу подключил?

по всей видимости нет

@tenKe за что?

меня тут забанили?

не, не тебя

Помадчина

вообще это шутка, конечно же

сча всех забаню (нет)

( :

я поправил

я тоже

хитрец какой

https://www.gridgain.com/resources/blog/apache-ignite-transactions-architecture-2-phase-commit-protocol

Привет! Кто может подсказать? Я раньше работал только с csv файлами в питоне. Сейчас у меня есть sequence file, и я не знаю как с ними работать. Их нужно конвертить в csv или какими-то средствами напрямую с ними работать?

Привет! Кто может подсказать? Я раньше работал только с csv файлами в питоне. Сейчас у меня есть sequence file, и я не знаю как с ними работать. Их нужно конвертить в csv или какими-то средствами напрямую с ними работать?

https://gist.github.com/samklr/15977b4abe82bd5d429e

а можно ли в airflow узнать результат предыдущего оператора? например, в результате действий оператора О1 создан файл, а оператору О2 надо этот файл как инпут

а можно ли в airflow узнать результат предыдущего оператора? например, в результате действий оператора О1 создан файл, а оператору О2 надо этот файл как инпут

Почитайте про airflow xcom. Может, это то, что нужно

Почитайте про airflow xcom. Может, это то, что нужно

гляну, спасибо

гляну, спасибо

А что за операторы?

А что за операторы?

часть BashOperator, часть PythonOperator

часть BashOperator, часть PythonOperator

Так а зачем вам как-то их передавать? Если уверены, что файл создался, то и обрабатывайте его следующим оператором. Вы же знаете путь к файлу.

Так а зачем вам как-то их передавать? Если уверены, что файл создался, то и обрабатывайте его следующим оператором. Вы же знаете путь к файлу.

с файлами-то да а если там, скажем, np.array, то уже, похоже, нужен xcom

или запиклить ?

с файлами-то да а если там, скажем, np.array, то уже, похоже, нужен xcom

Это да, но airflow не советует передавать большие объёмы через xcom. Учтите.

наверное, лучше будет как-то эти операторы слить, чтобы туда-сюда не гонять

наверное, лучше будет как-то эти операторы слить, чтобы туда-сюда не гонять

Ну можно в python операторе выполнить всё , что вам надо в bash через subprocess, например.

Ну можно в python операторе выполнить всё , что вам надо в bash через subprocess, например.

угу, так и сделаю а можно ли в операторе создать вызов динамического (в рантайме) количества других операторов?

угу, так и сделаю а можно ли в операторе создать вызов динамического (в рантайме) количества других операторов?

С import operator можно всё

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

угу, так и сделаю а можно ли в операторе создать вызов динамического (в рантайме) количества других операторов?

Оператор в операторе? Сомневаюсь. Тоже, если правильно понял, о чём речь. Но точно можно даг в даге.

Ааа, это что-то спарковское таки, я думал речь про перегрузку питоньих

Кстати, раз уж зашел сюда, у меня получилось возвращать датафрейм в питон из скала расширения и даже обернуть его в питоновский датафрейм так чтобы работать с ним seamlessly так сказать. Но есть нюанс с тем что весь этот велосипед надо запускать на драйвере, т.к. SparkSession в экзекутор не передать

Кстати, раз уж зашел сюда, у меня получилось возвращать датафрейм в питон из скала расширения и даже обернуть его в питоновский датафрейм так чтобы работать с ним seamlessly так сказать. Но есть нюанс с тем что весь этот велосипед надо запускать на драйвере, т.к. SparkSession в экзекутор не передать

а если сделать глобал вью ?

а если сделать глобал вью ?

Ммм, не пробовал

Если есть ссылка чтобы кинуть то почитаю

Ребят, вопрос из разряда про датафреймы. Вот допустим у нас есть uid:String, price:Long, timestamp:Long /*или как хотите*/ и нужно допустим сделать .withColumn("average_last_year", col("price").average.over(Window. ....... ) ) есть вменяемый способ для каждой строки посчитать среднее за ровно календарный год( с учётом там високосного и прочих прелестей ) или довольствоваться только визинрейджом ( с относительным смещением) и ордер баем внутри окна

def timestampBasedWindow(cols: Seq[String], timeStampCol: String, start: Double, end: Double): WindowSpec = { val secondsInYear = 60 * 60 * 24 * 365 val effStart = (start * secondsInYear).toLong val effEnd = (end * secondsInYear).toLong Window .partitionBy(cols.head, cols.tail: _*) .orderBy(timeStampCol) .rangeBetween(effStart, effEnd) }

Кстати, раз уж зашел сюда, у меня получилось возвращать датафрейм в питон из скала расширения и даже обернуть его в питоновский датафрейм так чтобы работать с ним seamlessly так сказать. Но есть нюанс с тем что весь этот велосипед надо запускать на драйвере, т.к. SparkSession в экзекутор не передать

На экзекуторы не стоит ее передавать вообще

На экзекуторы не стоит ее передавать вообще

Я и не планирую, думаю перенести нужную мне логику в скалу и всё что надо делать распределенно для задачи делать там

Никаких минусов и сплошные плюсы

Ну а репорты пусть пишут на питоне, датафрейм то вернётся с данными

Alexander а ещё такой вопрос насчет аирфлоу — вот у меня есть директория с 100000 файлов, я хочу все их запроцессить параллельно, как это указать в даге?

"запусти мне параллельную обработку всего в вооон той директории"

Alexander а ещё такой вопрос насчет аирфлоу — вот у меня есть директория с 100000 файлов, я хочу все их запроцессить параллельно, как это указать в даге?

Цикл по этим файлам сделать, а внутри цикла ваш оператор. Завтра на работу приду, скину вам пример, напомните только.

а, похоже понял, попробую, спасибо

продолжая тему с airflow — пайплайн на Sequential работает отлично, запускал через backfill переехал на Celery, почему-то не стартуют SubDagOperator вывод из test: [2018-01-19 09:58:59,417] {jobs.py:1978} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 0 | succeeded: 0 | kicked_off: 2 | failed: 0 | skipped: 0 | deadlocked: 0 | not ready: 0

запускал и через run 2018-01-19, и через backfill, и trigger_dag (отдельный вопрос по этому, он вообще ни разу не заработал, аутпута нет, хз чего делает вообще)

блин вроде уменя не было никаких проблем с написанием дагов

я дебажу принтлнами их, попробуй

блин вроде уменя не было никаких проблем с написанием дагов

Написанием дагов?

Не плохо

блин вроде уменя не было никаких проблем с написанием дагов

с написанием ок, с запуском туплю

'@once', 2018-01-19, run/backfill — не работает с Celery, но ок с Sequential

вот как вы запускаете, допустим, run-once даг?

офтоптик, мутабельные динамические недекларативные workflow - зло

@dveim а для чего вы используете airflow?

ну бизнес логика какая

@dveim а для чего вы используете airflow?

для эксперимента сейчас) есть набор файлов, логика процессинга одного, и хочу с airflow раскидать это все по воркерам на кластере

понятно, ок

А знает кто-нибудь, возможно ли в hive каким-то образом создавать талицы, в которых имена столбцов записываются в верхнем регистре, а не в нижнем?

с написанием ок, с запуском туплю

воркеров-то не запустил?

#airflow как правильнее всего запускать map-reduce операции? в основном интересует сборки инфы на редьюсере, мапы вроде как через subdag задаются xcom ?

Рибята по кафке вопросы; гоняю тесты локально на больших месажах стримы очень медленно просто получают данные stream.foreach { (key, value) => println(value) } куда тыкать? конечно это может быть проблема и ембедед кафки, проверю на днях)

Рибята по кафке вопросы; гоняю тесты локально на больших месажах стримы очень медленно просто получают данные stream.foreach { (key, value) => println(value) } куда тыкать? конечно это может быть проблема и ембедед кафки, проверю на днях)

https://t.me/proKafka

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

дефолтный размер батча у консамера вроде 100

для 5 метровых месаг будет уже по полгига

@optician_owl спасиб похоже те конфиги которые смотреть спс и за канал

товарищи, ни у кого не найдётся rpm с pig 0.16.0 под hadoop 2.6?

так не подойдет? https://search.maven.org/#artifactdetails|org.apache.pig|pig|0.16.0|jar

версия хадупа не указана, но я бы просто потестил

исходя из доки клаудеры http://www.cloudera.com/documentation/cdh/5-1-x/CDH5-Installation-Guide/cdh5ig_pig_install.html (кусок логов), есть гипотеза, что это в принципе одна джарка со скриптом запуска.

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

короче похоже ембедед слабенький

не оч мне понятно где все валится но может брокер просто пытатся выделить мног опамяти (или прост тугой)

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