pyspark sql函数而不是rdd distinct-Java 学习之路

我一直在尝试替换特定列的数据集中的字符串 . 1或0，'Y'如果为1，否则为0 .

我已经设法确定要使用lambda对数据帧进行rdd转换的目标列，但是需要一段时间来处理 .

每个列都会切换到rdd，然后执行不同的操作，这需要一段时间！

如果不同的结果集中存在“Y”，则该列被识别为需要转换 .

我想知道是否有人可以建议我如何专门使用pyspark sql函数来获得相同的结果而不必为每列切换？

样本数据的代码如下：

import pyspark.sql.types as typ
    import pyspark.sql.functions as func

    col_names = [
        ('ALIVE', typ.StringType()),
        ('AGE', typ.IntegerType()),
        ('CAGE', typ.IntegerType()),
        ('CNT1', typ.IntegerType()),
        ('CNT2', typ.IntegerType()),
        ('CNT3', typ.IntegerType()),
        ('HE', typ.IntegerType()),
        ('WE', typ.IntegerType()),
        ('WG', typ.IntegerType()),
        ('DBP', typ.StringType()),
        ('DBG', typ.StringType()),
        ('HT1', typ.StringType()),
        ('HT2', typ.StringType()),
        ('PREV', typ.StringType())
        ]

    schema = typ.StructType([typ.StructField(c[0], c[1], False) for c in col_names])
    df = spark.createDataFrame([('Y',22,56,4,3,65,180,198,18,'N','Y','N','N','N'),
                                ('N',38,79,3,4,63,155,167,12,'N','N','N','Y','N'),
                                ('Y',39,81,6,6,60,128,152,24,'N','N','N','N','Y')]
                               ,schema=schema)

    cols = [(col.name, col.dataType) for col in df.schema]

    transform_cols = []

    for s in cols:
      if s[1] == typ.StringType():
        distinct_result = df.select(s[0]).distinct().rdd.map(lambda row: row[0]).collect()
        if 'Y' in distinct_result:
          transform_cols.append(s[0])

    print(transform_cols)

输出是：

['ALIVE', 'DBG', 'HT2', 'PREV']

1 回答

我设法使用 udf 来完成任务 . 首先，选择 Y 或 N 列（这里我使用 func.first 以浏览第一行）：

cols_sel = df.select([func.first(col).alias(col) for col in df.columns]).collect()[0].asDict()
cols = [col_name for (col_name, v) in cols_sel.items() if v in ['Y', 'N']]
# return ['HT2', 'ALIVE', 'DBP', 'HT1', 'PREV', 'DBG']

接下来，您可以创建 udf 函数，以便将 Y ， N 映射到 1 ， 0 .

def map_input(val):
    map_dict = dict(zip(['Y', 'N'], [1, 0]))
    return map_dict.get(val)
udf_map_input = func.udf(map_input, returnType=typ.IntegerType())

for col in cols:
    df = df.withColumn(col, udf_map_input(col))
df.show()

最后，您可以对列进行求和 . 然后我将输出转换为字典并检查哪些列的值大于0（即包含 Y ）

out = df.select([func.sum(col).alias(col) for col in cols]).collect()
out = out[0]
print([col_name for (col_name, val) in out.asDict().items() if val > 0])

Output

['DBG', 'HT2', 'ALIVE', 'PREV']

回复于 2024-05-10T04:38:24+08:00

pyspark sql函数而不是rdd distinct

1 回答

相关问题