Spark - Как объединить текущие и предыдущие записи в DataFrame и назначить исходное поле для всех таких случаев - PullRequest
Новая
       8

Spark - Как объединить текущие и предыдущие записи в DataFrame и назначить исходное поле для всех таких случаев

0 голосов
/ 31 декабря 2018

Мне нужно просканировать таблицу Hive и добавить значения из первой записи в последовательности ко всем связанным записям.

Логика будет такой: -

  1. Найти первуюзапись (где предыдущий_ид пуст).
  2. Найти следующую запись (current_id = previous_id).
  3. Повторять до тех пор, пока не будет больше связанных записей.
  4. Добавить столбцы из исходной записи во все связанные записи.
  5. Вывод результатов в таблицу Hive.

Пример исходных данных: - 

current_id previous_id start_date
---------- ----------- ----------
100                    01/01/2001
200        100         02/02/2002
300        200         03/03/2003

Пример выходных данных: - 

current_id start_date
---------- ----------
100        01/01/2001
200        01/01/2001
300        01/01/2001

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

Есть предложения о том, как решить эту проблему?

Ответы [ 3 ]

0 голосов
/ 02 января 2019

Я думаю, что вы можете сделать это, используя GraphFrames Connected компоненты

Это поможет вам избежать написания логики контрольных точек и циклических операций самостоятельно.По сути, вы создаете график из пар current_id и previous_id и используете GraphFrames для компонента для каждой вершины.Полученный в результате DataFrame можно затем присоединить к исходному DataFrame, чтобы получить start_date. 

from graphframes import *

sc.setCheckpointDir("/tmp/chk")

input = spark.createDataFrame([
  (100, None, "2001-01-01"),
  (200, 100, "2002-02-02"),
  (300, 200, "2003-03-03"),
  (400, None, "2004-04-04"),
  (500, 400, "2005-05-05"),
  (600, 500, "2006-06-06"),
  (700, 300, "2007-07-07")
], ["current_id", "previous_id", "start_date"])

input.show()

vertices = input.select(input.current_id.alias("id"))

edges = input.select(input.current_id.alias("src"), input.previous_id.alias("dst"))

graph = GraphFrame(vertices, edges)

result = graph.connectedComponents()

result.join(input.previous_id.isNull(), result.component == input.current_id)\
  .select(result.id.alias("current_id"), input.start_date)\
  .orderBy("current_id")\
  .show()

. В результате получится следующий вывод: 

+----------+----------+
|current_id|start_date|
+----------+----------+
|       100|2001-01-01|
|       200|2001-01-01|
|       300|2001-01-01|
|       400|2004-04-04|
|       500|2004-04-04|
|       600|2004-04-04|
|       700|2001-01-01|
+----------+----------+
0 голосов
/ 18 января 2019

Спасибо за предложения, размещенные здесь.Попробовав различные подходы, я выбрал следующее решение, которое работает для нескольких итераций (например, 20 циклов) и не вызывает проблем с памятью.

«Физический план» все еще огромен, но кэширование означает, что большинство шагов пропущено, что обеспечивает приемлемую производительность. 

input = spark.createDataFrame([
    (100, None, '2001/01/01'),
    (200, 100,  '2002/02/02'),
    (300, 200,  '2003/03/03'),
    (400, None, '2005/01/01'),
    (500, 400,  '2006/02/02'),
    (600, 300,  '2007/02/02'),
    (700, 600,  '2008/02/02'),
    (800, None, '2009/02/02'),
    (900, 800,  '2010/02/02')
], ["current_id", "previous_id", "start_date"])

input.createOrReplaceTempView("input")

cur = spark.sql("select * from input where previous_id is null")
nxt = spark.sql("select * from input where previous_id is not null")

cur.cache()
nxt.cache()

cur.createOrReplaceTempView("cur0")
nxt.createOrReplaceTempView("nxt")

i = 1
while True:
    spark.sql("set table_name=cur" + str(i - 1))
    cur = spark.sql(
        """
            SELECT  nxt.current_id  as current_id,  
                    nxt.previous_id as previous_id, 
                    cur.start_date  as start_date       
            FROM    ${table_name}   cur, 
                    nxt             nxt 
            WHERE   cur.current_id = nxt.previous_id 
        """).cache()
    cur.createOrReplaceTempView("cur" + str(i))
    i = i + 1
    if cur.count() == 0:
        break

for x in range(0, i):
    spark.sql("set table_name=cur" + str(x))
    cur = spark.sql("select * from ${table_name}")
    if x == 0:
        out = cur
    else:
        out = out.union(cur)
0 голосов
/ 02 января 2019

Вот подход, который, я не уверен, хорошо подходит для Spark.

Отсутствует идентификатор / ключ группировки для данных.

Не уверен, как Catalyst сможет оптимизировать это - рассмотрим позже.Ошибки памяти, если слишком велики?

Сделали данные более сложными, и это работает.Здесь идет: 

# No grouping key evident, more a linked list with asc current_ids.
# Added more complexity to the example.
# Questions open on performance at scale. Interested to see how well Catalyst handles this.
# Need really some grouping id/key in the data.

from pyspark.sql import functions as f
from functools import reduce
from pyspark.sql import DataFrame
from pyspark.sql.functions import col

# Started from dataframe.
# Some more realistic data? At least more complex.
columns = ['current_id', 'previous_id', 'start_date']
vals = [
        (100, None, '2001/01/01'),
        (200, 100,  '2002/02/02'),
        (300, 200,  '2003/03/03'),
        (400, None, '2005/01/01'),
        (500, 400,  '2006/02/02'),
        (600, 300,  '2007/02/02'),
        (700, 600,  '2008/02/02'),
        (800, None, '2009/02/02'),
        (900, 800,  '2010/02/02')  
       ]
df = spark.createDataFrame(vals, columns)
df.createOrReplaceTempView("trans")

# Starting data. The null / None entries. 
df2 = spark.sql("""
                   select * 
                     from trans 
                    where previous_id is null
                """)
df2.cache
df2.createOrReplaceTempView("trans_0")

# Loop through the stuff based on traversing the list elements until exhaustion of data, and, write to dynamically named TempViews.
# May need to checkpoint? Depends on depth of chain of linked items.
# Spark not well suited to this type of processing.  
dfX_cnt  = 1
cnt = 1

while (dfX_cnt != 0): 
  tabname_prev = 'trans_' + str(cnt-1)
  tabname = 'trans_' + str(cnt) 

  query = "select t2.current_id, t2.previous_id, t1.start_date from {} t1, trans t2 where t1.current_id = t2.previous_id".format(tabname_prev)
  dfX = spark.sql(query)
  dfX.cache

  dfX_cnt = dfX.count()
  if (dfX_cnt!=0):
      #print('Looping for dynamic creation of TempViews')
      dfX.createOrReplaceTempView(tabname)
      cnt=cnt+1

# Reduce the TempViews all to one DF. Can reduce an array of DF's as well, but could not find my notes here in this regard.
# Will memory errors occur? 

from pyspark.sql.types import *
fields = [StructField('current_id', LongType(), False),
          StructField('previous_id', LongType(), True),
          StructField('start_date',  StringType(), False)]
schema = StructType(fields)
dfZ = spark.createDataFrame(sc.emptyRDD(), schema)

for i in range(0,cnt,1):
    tabname = 'trans_' + str(i)
    query = "select * from {}".format(tabname)
    df = spark.sql(query)
    dfZ = dfZ.union(df)

# Show final results.
dfZ.select('current_id', 'start_date').sort(col('current_id')).show()

возвращает: 

+----------+----------+
|current_id|start_date|
+----------+----------+
|       100|2001/01/01|
|       200|2001/01/01|
|       300|2001/01/01|
|       400|2005/01/01|
|       500|2005/01/01|
|       600|2001/01/01|
|       700|2001/01/01|
|       800|2009/02/02|
|       900|2009/02/02|
+----------+----------+
...