pyspark案例系列3-dataframe实现mysql的group_concat功能

蔚落 2022-10-15 07:00 135阅读 0赞

# 一.问题描述 #

今天写pyspark遇到一个问题，要实现同mysql的GROUP\_CONCAT函数的功能

数据1:

col1   col2
    1        a
    1        b
    1        c
    2        d
    2        f

想要的结果1:

col1   new_col2
    1        a,b,c
    2        d,f

如果存在多列是否也可行  
数据2:

col1   col2   col3
    1        a        100
    1        b        200
    1        c        300
    2        d        400
    2        f        500

想要的结果2:

col1   new_col2    new_col3
    1        a,b,c   100,200,300
    2        d,f       400,500

# 二.解决方案 #

pyspark的collect\_list可以实现如下需求

代码：

#!/usr/bin/env python
    from pyspark import SparkContext, SparkConf
    from pyspark.sql.session import SparkSession
    from pyspark.sql.functions import collect_list
    from pyspark.sql.functions import collect_set
    
    spark_conf = SparkConf().setMaster("local").setAppName("test1")
    sc = SparkContext(conf = spark_conf)
    spark = SparkSession(sc)
    
    df1 = spark.createDataFrame([('1','a'),('1','b'),('1','c'),('2','d'),('2','f')], ['col1', 'col2'])
    df1.groupBy("col1").agg(collect_list('col2').alias('new_col2')).show()
    
    df2 = spark.createDataFrame([('1','a','100'),('1','b','200'),('1','c','300'),('2','d','400'),('2','f','500')], ['col1', 'col2', 'col3'])
    df2.groupBy("col1").agg(*[collect_set(col) for col in ['col2','col3']]).show()
    
    sc.stop()

测试记录:

--snip--
    21/04/15 17:26:41 INFO hive.metastore: Trying to connect to metastore with URI thrift://hp1:9083
    21/04/15 17:26:41 INFO hive.metastore: Opened a connection to metastore, current connections: 1
    21/04/15 17:26:41 INFO hive.metastore: Connected to metastore.
    +----+---------+
    |col1| new_col2|
    +----+---------+
    |   1|[a, b, c]|
    |   2|   [d, f]|
    +----+---------+
    --snip--
    21/04/15 17:30:43 INFO scheduler.TaskSetManager: Finished task 64.0 in stage 19.0 (TID 401) in 8 ms on localhost (executor driver) (75/75)
    21/04/15 17:30:43 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 19.0, whose tasks have all completed, from pool 
    21/04/15 17:30:43 INFO scheduler.DAGScheduler: ResultStage 19 (showString at NativeMethodAccessorImpl.java:0) finished in 0.272 s
    21/04/15 17:30:43 INFO scheduler.DAGScheduler: Job 9 finished: showString at NativeMethodAccessorImpl.java:0, took 0.278859 s
    +----+-----------------+-----------------+
    |col1|collect_set(col2)|collect_set(col3)|
    +----+-----------------+-----------------+
    |   1|        [c, b, a]|  [100, 300, 200]|
    |   2|           [f, d]|       [500, 400]|
    +----+-----------------+-----------------+
    --snip--

参考:  
1.https://www.cnblogs.com/TTyb/p/10196544.html