Consider you have a Dataframe like below +—–+——+—+ | name|sector|age| +—–+——+—+ | Andy| aaa| 20| |Berta| bbb| 30| | Joe| ccc| 40| +—–+——+—+ To loop your Dataframe and extract the elements from the Dataframe, you can either chose one of the below approaches. Approach 1 – Loop using foreach Looping a dataframe directly using foreach … Read more
Apache Spark does not support native CSV output on disk. You have four available solutions though: You can convert your Dataframe into an RDD : def convertToReadableString(r : Row) = ??? df.rdd.map{ convertToReadableString }.saveAsTextFile(filepath) This will create a folder filepath. Under the file path, you’ll find partitions files (e.g part-000*) What I usually do if … Read more
from pyspark.sql.functions import concat, col, lit df.select(concat(col(“firstname”), lit(” “), col(“lastname”))).show(5) +——————————+ |concat(firstname, , lastname)| +——————————+ | Emanuel Panton| | Eloisa Cayouette| | Cathi Prins| | Mitchel Mozdzierz| | Angla Hartzheim| +——————————+ only showing top 5 rows http://spark.apache.org/docs/2.0.0/api/python/pyspark.sql.html#module-pyspark.sql.functions
the following code is not working With single element you need a schema as type spark.createDataFrame([“10″,”11″,”13”], “string”).toDF(“age”) or DataType: from pyspark.sql.types import StringType spark.createDataFrame([“10″,”11″,”13”], StringType()).toDF(“age”) With name elements should be tuples and schema as sequence: spark.createDataFrame([(“10”, ), (“11”, ), (“13”, )], [“age”])
This person was having similar issue: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-use-DataFrame-with-MySQL-td22178.html Have you updated your connector drivers to the most recent version? Also did you specify the driver class when you called load()? Map<String, String> options = new HashMap<String, String>(); options.put(“url”, “jdbc:mysql://localhost:3306/video_rcmd?user=root&password=123456”); options.put(“dbtable”, “video”); options.put(“driver”, “com.mysql.cj.jdbc.Driver”); //here DataFrame jdbcDF = sqlContext.load(“jdbc”, options); In spark/conf/spark-defaults.conf, you can also set spark.driver.extraClassPath … Read more
Use either drop with subset: df.na.drop(subset=[“col_X”]) or isNotNull() df.filter(df.col_X.isNotNull())
Spark 2.0 This is issue has been resolved in Spark 2.0 with SPARK-14241. Another similar issue has been resolved in Spark 2.1 with SPARK-14393 Spark 1.x Problem you experience is rather subtle but can be reduced to a simple fact monotonically_increasing_id is an extremely ugly function. It is clearly not pure and its value depends … Read more
Yes, it is “smart enough“. groupBy performed on a DataFrame is not the same operation as groupBy performed on a plain RDD. In a scenario you’ve described there is no need to move raw data at all. Let’s create a small example to illustrate that: val df = sc.parallelize(Seq( (“a”, “foo”, 1), (“a”, “foo”, 3), … Read more
For non-numeric but Orderable types you can use agg with max directly: from pyspark.sql.functions import col, max as max_ df = sc.parallelize([ (“2016-04-06 16:36”, 1234, 111, 1), (“2016-04-06 17:35”, 1234, 111, 5), ]).toDF([“datetime”, “userId”, “memberId”, “value”]) (df.withColumn(“datetime”, col(“datetime”).cast(“timestamp”)) .groupBy(“userId”, “memberId”) .agg(max_(“datetime”))) ## +——+——–+——————–+ ## |userId|memberId| max(datetime)| ## +——+——–+——————–+ ## | 1234| 111|2016-04-06 17:35:…| ## +——+——–+——————–+
What you need is the DataFrame aggregation function countDistinct: import sqlContext.implicits._ import org.apache.spark.sql.functions._ case class Log(page: String, visitor: String) val logs = data.map(p => Log(p._1,p._2)) .toDF() val result = logs.select(“page”,”visitor”) .groupBy(‘page) .agg(‘page, countDistinct(‘visitor)) result.foreach(println)