How can I filter `filter(lambda x:len(x[1])>=2)` i

问题:

I am not sure about how to filter(lambda x:len(x[1])>=2) in dataframe. I would like to improve the speed of my spark app. Thanks for your help!

This some context from my spark app:

article_ids = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="article_by_created_at", keyspace=source).load().where(range_expr).select('article','created_at').repartition(64*2)

axes = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="axes", keyspace=source).load()
speed_df = article_ids.join(axes,article_ids.article==axes.article).select(axes.article,axes.at,axes.comments,axes.likes,axes.reads,axes.shares) \
     .map(lambda x:(x.article,[x])).reduceByKey(lambda x,y:x+y) \
     .map(lambda x:(x[0],sorted(x[1],key=lambda y:y.at,reverse = False))) \
     .filter(lambda x:len(x[1])>=2) \
     .map(lambda x:x[1][-1]) \
     .map(lambda x:(x.article,(x,(x.comments if x.comments else 0)+(x.likes if x.likes else 0)+(x.reads if x.reads else 0)+(x.shares if x.shares else 0))))

EDIT

I tried editing this but there was no speed improvement and the result was the same as before, did I do something wrong?:

axes = sqlContext.read.format("org.apache.spark.sql.cassandra").options(table="axes", keyspace=source).load()
        axes_filter_2rows=axes.groupBy("article").agg({"*": "count"}).where("COUNT(1)>=2")#.select('article',col("COUNT(1)").alias('count'))
        axes_max = axes.groupBy("article").max('comments','reads','likes','shares').select('article',col("MAX(comments)").alias('comments'),col("MAX(likes)").alias('likes'),col("MAX(reads)").alias('reads'),col("MAX(shares)").alias('shares'))
        axes_max_filter=axes_filter_2rows.join(axes_max,axes_filter_2rows.article==axes_max.article).select(axes_max.article,axes_max.comments,axes_max.likes,axes_max.reads,axes_max.shares)
        speed_df = article_ids.join(axes_max_filter,article_ids.article==axes.article).select(axes_max_filter.article,axes_max_filter.comments,axes_max_filter.likes,axes_max_filter.reads,axes_max_filter.shares) \
                    .map(lambda x:(x.article,(x,(x.comments if x.comments else 0)+(x.likes if x.likes else 0)+(x.reads if x.reads else 0)+(x.shares if x.shares else 0))))

回答1:

If you're just looking to filter rows in your dataframe that are greater than or equal to two, you can use an index:

speed_df = speed_df[speed_df.column >= 2]