Спарк с синтаксисом sql

А что значит "sql работает" тогда?

Запрос на реляционке отрабатывает на тех же данных

Спарк не поддерживает подзапрос в запросе, мне кажется мы неделю назад это обсуждали

Диалекты разные

Вам скорее всего дорога в cte

Молоко попробовать спросить у sqlglot

Да, я хотел убедиться, что то, что я пытаюсь сделать не получится данным способом

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

Вот честно, на мой взгляд, выглядит максимально непонятно.
У меня, мб, профдеформация какая-нибудь, но если это разложить на несколько аккуратных блоков dataframe/dataset API, то будет в сотню раз понятнее и удобнее, особенно в переиспользовании.

кажется стоит добавить бота-фильтратора, а то в последнее время часто спам прилетает какой-то

Да норм, мы держим оборону небыстро

Все привет, кто-нибудь сталкивался с проблемами с udf в 3 спарке
https://pastebin.com/eVG4pMk2

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

Всем привет! Столкнулся с неизвестной мне ошибкой на EMR кластере, который прожил 3 часа и внезапно ушел в кому 🙁

21/09/14 11:28:01 INFO TransportClientFactory: Successfully created connection to ip-10-0-2-232.eu-west-1.compute.internal/10.0.2.232:34145 after 93 ms (0 ms spent in bootstraps)
21/09/14 11:28:29 WARN TransportChannelHandler: Exception in connection from ip-10-0-2-232.eu-west-1.compute.internal/10.0.2.232:34145
java.io.IOException: Connection reset by peer
 at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
 at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
 at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
 at sun.nio.ch.IOUtil.read(IOUtil.java:192)
 at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:377)
 at io.netty.buffer.PooledUnsafeDirectByteBuf.setBytes(PooledUnsafeDirectByteBuf.java:288)
 at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1106)
 at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:343)
 at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123)
 at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
 at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
 at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
 at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
 at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
 at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
 at java.lang.Thread.run(Thread.java:748)

Причем, предыдущие job-ы на этом же самом кластере были успешные, а вот последняя в коме 🙁 Может быть кто-то сталкивался с чем-то похожим и подскажет куда копать? (Переезд на k8s в процессе, только EMR пока еще пользуется популярностью в компании 🙂 )

Заранее спасибо!

Привет, кто нибудь знает как можно разбить на строки, где признак новой строки значение null

+---+--------------+--------------------+
| id|           key|               value|
+---+--------------+--------------------+
|  0|              |                null|
|  1|            as|                  v1|
|  2|         descr|                  v2|
|  3|         descr|                  v3|
|  4|              |                null|

Результат:

+--------------+--------------------+
|            as|               descr|
+--------------+--------------------+
|           v1 |               v2,v3|

1. перенумеровать value = null строки
например when(value.isNull(), F.lit(1)).otherwise(F.lit(None)), потом sum
2. поставить ранг остальным строкам функцией last(rank_column, ignorenulls=True)
3. можно делать group by

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

Всем привет, есть задача считать sum, min, max, stddev по нескольким колонкам. В пандасе это решается простым axis=1. В pyspark как я понимаю аналогичного решения нет и кроме как использовать питоновские функции в конструкциях типа .withColumn('SUM1',sum([F.col(c) for c in col_list])) выхода не вижу (что достаточно плохо скажется на производительности). Может быть есть какое-то более тру решение?

Сумма - это агрегатная функция, она исполняется по колонке по всем рядам, а не по нескольким колонкам

Если нужно по нескольким колонкам посчитать сумму, по нескольким максимум, минимум и стд, то можно сделать всё с помощью select