54767

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

Question:

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))))

<strong>EDIT</strong>

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))))

Answer1:

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]

Recommend

  • Parameterized queries with the Python Cassandra Module
  • Laravel and redis scan
  • Restore cassandra cluster data when acccidentally drop table
  • How implement LEFT or RIGHT JOIN using spark-cassandra-connector
  • how to implement fixed number of (timeuuid) columns in cassandra (with CQL)?
  • ('Unable to complete the operation against any hosts', {})
  • Cassandra: Adding new column to the table
  • How to reset Cassandra superuser, when Cassandra does not know 'cassandra' default user?
  • How do I store unsigned integers in Cassandra?
  • Java Caching frameworks for maintaining huge data
  • How to load more than one div at a time
  • How can Delete be both a DDL and a DML statement
  • How to 'create temp table as select' in Slick?
  • cordova is not defined - cordova.js has already been loaded :: Ionic
  • Azure webjobs output logs indexing taking very long
  • Best way to dynamically load an xml configuration file into a Flex4/Flash movie at runtime?
  • perl, mysql - fasting way to upload a csv file into mysql?
  • pillow imaging ImportError
  • C# program and C++ DLL compiled for 32-bit system crash on 64-bit system
  • pyodbc doesn't report sql server error
  • Loading .coffee files via a view in Rails
  • Create DicomImage from scratch using Dcmtk
  • PHPUnit_Framework_TestCase class is not available. Fix… - Makegood , Eclipse
  • Spring security and special characters
  • Nant, Vault & Windows Integrated Authentication
  • Deleting and Updating values from a cusrsor adapter
  • PHP - How to update data to MySQL when click a radio button
  • Counter field in MS Access, how to generate?
  • Submit form in a displaytag pagination
  • Modifying destination and filename of gulp-svg-sprite
  • 'TypeError' while using NSGA2 to solve Multi-objective prob. from pyopt-sparse in OpenMDAO
  • Javascript + PHP Encryption with pidCrypt
  • Websockets service method fails during R startup
  • Delete MySQLi record without showing the id in the URL
  • Unanticipated behavior
  • Comma separated Values
  • NSLayoutConstraint that would pin a view to the bottom edge of a superview
  • Why can't I rebase on to an ancestor of source changesets if on a different branch?
  • Trying to get generic when generic is not available
  • How to load view controller without button in storyboard?