PySpark Быстрый метод для разворота данных - PullRequest
Новая
       4

PySpark Быстрый метод для разворота данных

0 голосов
/ 21 сентября 2018

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

Метод 1: 

def to_long(df, by):

  # Filter dtypes and split into column names and type description
  cols, dtypes = zip(*((c, t) for (c, t) in df.dtypes if c not in by))
  # Spark SQL supports only homogeneous columns
  assert len(set(dtypes)) == 1, "All columns have to be of the same type"

  # Create and explode an array of (column_name, column_value) structs
  kvs = explode(array([
  struct(lit(c).alias("question_id"), col(c).alias("response_value")) for c in cols
])).alias("kvs")

return df.select(by + [kvs]).select(by + ["kvs.question_id", "kvs.response_value"])

Метод 2: 

def rowExpander(row):
  rowDict = row.asDict()
  valA = rowDict.pop('user_id')
  for k in rowDict:
     yield Row(**{'user_id': valA , 'question_id' : k, 'response_value' : row[k]})

user_response_df = spark.createDataFrame(response_df.rdd.flatMap(rowExpander))

Ответы [ 2 ]

0 голосов
/ 24 июля 2019

df.selectExpr('col1', 'stack(2, "col2", col2, "col3", col3) as (cols, values)')

0 голосов
/ 21 сентября 2018

Может быть, вы можете попробовать выбрать каждый столбец как новый фрейм данных и объединить их все
Как это 

# Get all columns except 'user_id'
cols = [col for col in df.columns if col != 'user_id']

# Select user_id and another column as a new dataframe.
# Use column_name as the value of the new column `question_id`
# Use column_value as the value of the new column `response_value`
# Then union all of these new dataframes
df = reduce(lambda df1, df2: df1.union(df2),
            [df.select('user_id',
                       F.lit(c).alias('question_id'), 
                       F.col(c).alias('response_value')) for c in cols])
...