А еще глупый вопрос) но я искал и как-то не нашел Есть возможность в спарковском датафрейме сделать что-то типа индекса по полю или как-то партиционировать? Для более быстрой работы с одним/несколькими полями?

кастомные партиции можно, индекс врядли

Ты для чего json на hdfs кидаешь? И зачем вообще json кладёшь? Даже если ты спарком будешь что-то агрегировать, тебе в какой-то момент потребуется либо много памяти либо, скидывать промежуточный итог на hdfs.

кастомные партиции можно, индекс врядли

Ну, индекс можно и самому наколхозить - делаем репартишен по полю с кастомным индексом, кешируем, а дальше фильтруем where spark_partition_id = get_index(key)

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

Господа, доборый день, вопросы по kafka тут будут уместны?

Господа, доборый день, вопросы по kafka тут будут уместны?

да, задавай

Отлично, вопрос следующий. Из кафки я получил сообщение, но не отправил ack. Пока я обрабатывал сообщение мне пришло новое сообщение с этим же ключем. Каковы в этом случае действия кафки?

Для кафки это просто новое сообщение

почему? ведь там же есть механизм дедуплекации

Ключ может использоваться твоим приложением и механизмом дедупликации

Господа, доборый день, вопросы по kafka тут будут уместны?

@proKafka

Дедупликация - это опциональная настройка топика, по умолчанию не применяется

у меня она включена

у меня она включена

Процесс дедубликации не гарантирует отсутствия дубликатов

Отлично, вопрос следующий. Из кафки я получил сообщение, но не отправил ack. Пока я обрабатывал сообщение мне пришло новое сообщение с этим же ключем. Каковы в этом случае действия кафки?

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

ну потому что я сообщение из кафки должен записать в эластик

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

Потому что хочет обработать батч, например

а в эластик оно может не записаться

через кафку приходят записи, которые были обновлены. Поэтому я хочу записывать только свежие сообщения. Для этого я использую дедупликацию. Но проблема в том, что если одна запись долго не записывается, а ack не послан, то в это время может прийти обновленная эта же запись и мне непонятно поведение кафки в этом случае.

Процесс дедубликации не гарантирует отсутствия дубликатов

хм, а что она тогда гарантирует?

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

хм, а что она тогда гарантирует?

Что с некоторой задержкой они будут подчищаться

Отлично, вопрос следующий. Из кафки я получил сообщение, но не отправил ack. Пока я обрабатывал сообщение мне пришло новое сообщение с этим же ключем. Каковы в этом случае действия кафки?

а если включить exactly-once, он вроде как раз для этого и предназначен

а если включить exactly-once, он вроде как раз для этого и предназначен

хм, спасибо, сейчас гляну что это

Примерно так. Здесь уже добавил repartition и cache

осталось теперь понять, где тормозит , на какой стадии, это не BatchPythonEvaluation?

а если включить exactly-once, он вроде как раз для этого и предназначен

Нет, он с логическими ключами не связан. Exactly once это доставка без дубликатов из одного топика в другой внутри кафки

Нет, он с логическими ключами не связан. Exactly once это доставка без дубликатов из одного топика в другой внутри кафки

нашел https://habr.com/company/badoo/blog/333046/ обсуждали это, пришли ко мнению что: Механизм exactly-once — он только для producer-ов, то есть если producer по какой-то причине не уверен, сохранилось ли сообщение в кафку (например разрыв соединения), он просто перепошлёт сообщение в кафку с тем же ID и механизм exactly-once гарантирует, что event не будет продублирован

зависит от нагрузки, сколько ядер приходится, сколько данных (партиций) на экзекутере и т.д.

и хватит ли памяти на каждый экзекютер

я выделил 30 экзекъюторов по 6 гигов и 2 ядра пошло побыстрее В теории таких экзекъюторов можно сделать и 100+, но мне как-то стремно отъедать все ресурсы кластера на одну задачу.

я выделил 30 экзекъюторов по 6 гигов и 2 ядра пошло побыстрее В теории таких экзекъюторов можно сделать и 100+, но мне как-то стремно отъедать все ресурсы кластера на одну задачу.

Попробуй сделать 60 по 2 гига

я выделил 30 экзекъюторов по 6 гигов и 2 ядра пошло побыстрее В теории таких экзекъюторов можно сделать и 100+, но мне как-то стремно отъедать все ресурсы кластера на одну задачу.

Почему стремно?

я выделил 30 экзекъюторов по 6 гигов и 2 ядра пошло побыстрее В теории таких экзекъюторов можно сделать и 100+, но мне как-то стремно отъедать все ресурсы кластера на одну задачу.

Используйте выделенную квоту и отдайте ее

Почему стремно?

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

Используйте выделенную квоту и отдайте ее

dynamic allocation в смысле?

dynamic allocation в смысле?

Какой RM используется для выполнения? YARN, Mesos и т.п.

Yarn

в клиент моде

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

--master yarn --deploy-mode client --executor-memory 4g --executor-cores 2 --conf spark.ui.enabled=true --queue airflow-py-spark --conf spark.shuffle.service.enabled=true --conf spark.dynamicAllocation.enabled=true --conf spark.dynamicAllocation.maxExecutors=60 С такими опциями запускаю

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

Очередь при этом может раздуваться, если остальные свободны

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

А если ресурсы заняты другой очередью и в ней еще много pending applications, то, получается, задача из другой очереди запустится раньше, чем задачи в соседней очереди при освобождении ресурсов?

Очередь при этом может раздуваться, если остальные свободны

Для очереди ты задаешь квоты, чтобы она не отбирала у соседей ресурсы.

Для очереди ты задаешь квоты, чтобы она не отбирала у соседей ресурсы.

Ну max capacity то ты можешь задавать хоть 146%.

Попробуй сделать 60 по 2 гига

попробовал 4, уже на них падает с нехваткой памяти :(

А если ресурсы заняты другой очередью и в ней еще много pending applications, то, получается, задача из другой очереди запустится раньше, чем задачи в соседней очереди при освобождении ресурсов?

Нет. Есть механизм preemption. Он отберёт ресурсы, чтобы обеспечить min capacity соседней очереди.

Часто даже в ходе исполнения в очереди 1.

попробовал 4, уже на них падает с нехваткой памяти :(

Так у тебя количество контейнеров вырастает. Ради интереса запусти с динамик аллок и посмотри что там выделится

без ограничения сверхну?

Можно и без ограничения. Для отладки этого хватит

Смотри скорость выполнения кода. Может получится так, что увеличение контейнеров не даёт никакого эффекта. Тогда ты упёрся в драйаео

Драйвер

Ну max capacity то ты можешь задавать хоть 146%.

Смысла нет в такой настройке) Речь о том, как поделить кластер между нагрузками.

Смысла нет в такой настройке) Речь о том, как поделить кластер между нагрузками.

вообще то имеет

Wat

Почему не имеет смысла ?

когда ты не можешь предсказать когда на какой очереди будет больше нагрузка…

когда ты не можешь предсказать когда на какой очереди будет больше нагрузка…

В данном случае ты хочешь гарантировать, что часть кластера останется кому-то еще. После этого Spark получает ресурсы в рамках очереди. В общем случае, ты можешь выставишь повышающий коэффициент, как сказал Рамиль. Если ты хочешь дальше нарезать на вложенные очереди, то там пожалуйста. Хотя можно пользоваться эластичностью.

E3-1230V6 под хадуп через blade кто пробовал?

E3-1230V6 под хадуп через blade кто пробовал?

Ты делаешь херню с блейдами

Ты делаешь херню с блейдами

это у меня программисты хотят сделать блейды с 2 hdd и типа рейзер вставить с m2 и процы 1230v6

я вот пытаюсь отговорить

как мне заявили у 8 1230v6 будет больше ядер и выше частота чем у gold 5120

Ищу экспертов на образовательную программу по дата инжинирингу, которые могли бы рассказать, как готовить под разные кейсы следующие инструменты: - Sphinx - PostgreSQL - Sqoop - Druid - Flink - Lambda и kappa-архитектура (вводное занятие). Может быть, здесь есть желающие. Преподавание оплачивается. Условия, подробности в личку.

как мне заявили у 8 1230v6 будет больше ядер и выше частота чем у gold 5120

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

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

там просто довольно большая задача по аналитике и хранению данных для аналитики

программисты предлагают 64 гб оперативы 1230v6, и 2 hdd на 3,5 и 7200 на 10 тб

и это делать блейдами

сначала 4 тб хотят протестить а лучше сразу 10

я же говорю что это всё идиотизм, надо 2,5 сас диски по 10к рпм и 1,8 тб

с голдами 5120 и 256 гб оперативы

V6 можно взять 20 штук за цену вашего голда одного)

Нужно БОЛЬШЕ VCORE

Я за программистов?

Фуфло эти сас диски, если всё упирается в то, что в ярне не хватает коров)

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

Тогда надо брать процы от амд

Не, ну не до такой же степени)

Хотя не, 2 диска по 10тб - это грустно, лучше хотя бы 6 дисков по 4тб

https://blog.selectel.com/amd-vs-intel-comparative-benchmarks/ тут по тестам не так все плохо у амд

Не, ну не до такой же степени)

Жизненный опыт?

Жизненный опыт?

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