Функция карты Spark Dataframe

val df1 = Seq(("Brian", 29, "0-A-1234")).toDF("name", "age", "client-ID")
val df2 = Seq(("1234", 555-5555, "1234 anystreet")).toDF("office-ID", "BusinessNumber", "Address")

Я пытаюсь запустить функцию в каждой строке кадра данных (в потоковом режиме). Эта функция будет содержать комбинацию кода Scala и API-кода Spark dataframe. например, я хочу взять 3 функции из df и использовать их для фильтрации второго фрейма данных с именем df2. Насколько я понимаю, UDF не может этого сделать. Теперь у меня весь код фильтрации работает просто отлично, без возможности применить его к каждой строке df.

Моя цель - сделать что-то вроде

df.select("ID","preferences").map(row => ( //filter df2 using row(0), row(1) and row(3) ))

Фреймы данных не могут быть объединены, между ними нет соединяемых отношений.

Хотя я использую Scala, ответ на Java или Python, вероятно, подойдет.

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


apache-spark apache-spark-sql spark-streaming apache-spark-dataset spark-dataframe
person Brian    schedule 13.04.2018    source источник
comment
мы до сих пор не знаем, что это был за случай? Оба стрима или что?   -  person eliasah    schedule 13.04.2018
comment
оба являются потоковыми кадрами данных, поступающими из тем кафки.   -  person Brian    schedule 13.04.2018

Ответы (1)


arrow_upward
1
arrow_downward

Насколько я понимаю, UDF не может этого сделать.

Это правильно, но также не может map (local Datasets кажется исключением Почему этот код Spark создает исключение NullPointerException?). Вложенная логика, подобная этой, может быть выражена только с помощью joins:

  • #P3# <блочная цитата> #P4# #P5#

  • Если один Dataset не streaming, вы можете переборщить с crossJoin, за которым следует filter, но это, конечно, вряд ли рекомендуется.

person Alper t. Turker    schedule 13.04.2018
comment
Я даже не знаю, как вы можете определить предикат для другого фрейма данных, не имея производного набора ключей. Это похоже на решение типа crossJoin... - person eliasah; 13.04.2018
comment
Я даже не знаю, потому что в OP пропущена filter логика. Но если вы можете фильтровать, вы всегда можете получить его, даже если он фиктивный (1 = 1). Думайте о LSH здесь как о представлении этого шаблона... :) - person Alper t. Turker; 13.04.2018
comment
хорошо, большое спасибо, я подумаю об этом, и когда у меня будет свое решение, я опубликую его как обновление. - person Brian; 13.04.2018
comment
@user9613318 user9613318 Я делаю это постоянно :) - person eliasah; 13.04.2018
comment
есть ли способ взять переменные из df1 и сохранить их в переменную scala.. т.е. val name = df1.name в потоковом контексте? - person Brian; 13.04.2018
comment
Вы имеете в виду что-то вроде collect? Есть приемник памяти, но я не думаю, что он принесет вам много пользы. - person Alper t. Turker; 13.04.2018
comment
поэтому я последовал предложению о присоединении. Я использовал перекрестное соединение. В моем случае первый фрейм данных должен иметь только 1 строку. Если их больше, то это может быть действительно плохой путь. Еще одно решение, о котором я подумал, заключалось в том, чтобы встроить каждую строку df1 в виде массива в элемент столбца в df2. Я считаю, что это будет лучше масштабироваться. - person Brian; 13.04.2018
comment
@Brian crossJoin на самом деле плохо масштабируется. Он имеет сложность O (N ^ 2). То, что вы делаете, немного сбивает с толку, потому что, если оба ваших DF передаются в потоковом режиме, вы не должны были следовать первому предложению с эквивалентными соединениями, а не с перекрестными соединениями... - person eliasah; 13.04.2018
comment
@eliasah позвольте мне сделать резервную копию, я мог бы сделать это и в потоковом режиме, но я, вероятно, не буду по другим причинам. df1 однострочный кадр данных будет передаваться из темы kafka, df2, скорее всего, будет df из таблицы из Hive, которая обновляется, чтобы оставаться в курсе. Я знаю, что перекрестные соединения не масштабируются. Что я хочу сделать, так это отфильтровать df2, используя значения в df. - person Brian; 13.04.2018
comment
@eliasah один фильтр должен был бы взять значение 1 столбца в df и найти все строки в df, которые равны ‹=, другой фильтр сделает что-то подобное и т. Д. Есть несколько слоев фильтров. - person Brian; 13.04.2018