Спецификация байт-кода Java [закрывается]

Вот родной подход Spark, который не затрудняет имена столбцов. Он основан на aggregateByKey и использует словарь для сбора столбцов, которые отображаются для каждого ключа. Затем мы собираем все имена столбцов, чтобы создать окончательный файл данных. [Предварительная версия использовала jsonRDD после испускания словаря для каждой записи, но это более эффективно.] Ограничение на конкретный список столбцов или исключение таких, как XX, было бы легкой модификацией.

Производительность кажется хорошим даже на довольно больших столах. Я использую вариацию, которая подсчитывает количество раз, каждое из которых имеет переменное число событий для каждого идентификатора, генерируя один столбец для каждого типа события. Код в основном тот же, за исключением того, что для подсчета вхождений используется коллекция.Counter вместо dict в seqFn.

from pyspark.sql.types import *

rdd = sc.parallelize([('X01',41,'US',3),
                       ('X01',41,'UK',1),
                       ('X01',41,'CA',2),
                       ('X02',72,'US',4),
                       ('X02',72,'UK',6),
                       ('X02',72,'CA',7),
                       ('X02',72,'XX',8)])

schema = StructType([StructField('ID', StringType(), True),
                     StructField('Age', IntegerType(), True),
                     StructField('Country', StringType(), True),
                     StructField('Score', IntegerType(), True)])

df = sqlCtx.createDataFrame(rdd, schema)

def seqPivot(u, v):
    if not u:
        u = {}
    u[v.Country] = v.Score
    return u

def cmbPivot(u1, u2):
    u1.update(u2)
    return u1

pivot = (
    df
    .rdd
    .keyBy(lambda row: row.ID)
    .aggregateByKey(None, seqPivot, cmbPivot)
)
columns = (
    pivot
    .values()
    .map(lambda u: set(u.keys()))
    .reduce(lambda s,t: s.union(t))
)
result = sqlCtx.createDataFrame(
    pivot
    .map(lambda (k, u): [k] + [u.get(c) for c in columns]),
    schema=StructType(
        [StructField('ID', StringType())] + 
        [StructField(c, IntegerType()) for c in columns]
    )
)
result.show()

Производит:

ID  CA UK US XX  
X02 7  6  4  8   
X01 2  1  3  null