Pyspark RDD 在值中消除 None？

Question

我是 pyspark RDD 的新手，并且有一个从 JSON 文件中获取的数据框：

行(created_at='2021-05-05 23:37:51', hash_tags=None, id=1390088382659895296, replyto_id=None, replyto_user_id=None, retweet_id=1390027514332991489, retweet_user_id=807095, 文本='RT @nytimes: Breaking News: The Biden administration will support lifting patent protections for Covid-19 vaccines, a breakthrough for glob…', user_id=17799542, user_mentions=[行(id=807095, indices=[3, 11])])

这是我的全部代码：

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()


data_rdd = spark.read.option("multiline","true")\
    .json("tweets.json")
 
print(data_rdd.collect()[0])

def extractColumns(record):
    return (record[8],[record[4], record[6]])

ddata_frame = data_rdd.rdd.map(extractColumns)\
    .groupByKey()\
    .map(lambda r: (r[0], list(r[1])))

我以以下形式获得了RDD数据： [(17799542, [[None, 807095]]),

...

(3094649957, [[None, 3094649957], [None, None], [None, 3094649957], [None, None], [None, 3094649957], [None, None]])]

如何消除值中的 None 以实现以下目标： [(17799542, [807095]),

...

(3094649957, [3094649957, 3094649957, 3094649957])]

我试过以下但没有用：

def eliminateNone(record):
    s = list(filter(lambda s: each != None for each in s))
    return (record[0], s)

data_frame.mapValues(eliminateNone)
print(data_frame.collect())

非常感谢您的帮助。

Answer 1

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()


data_rdd = spark.read.option("multiline","true")\
    .json("tweets.json")


 
print(data_rdd.collect()[0])

def extractColumns(record):
    return (record[8],[record[3], record[5]])

def merge_values(data):
    result = []
    for l in data:
        for x in l:
            if x != None:
                result.append(x)
    return result

data_frame = data_rdd.rdd.map(extractColumns)\
    .groupByKey()\
    .map(lambda r: (r[0], list(r[1]))) 


data_frame = data_frame.mapValues(merge_values)
print(data_frame.collect())

你可以试试这个。

Pyspark RDD 在值中消除 None？

Pyspark RDD Eliminate None in values?

pyspark

apache-spark-sql

rdd