Write your first Apache Spark job

Write your first Apache Spark job

To write your first Apache Spark job, you add code to the cells of a Azure Databricks notebook. This example uses Python. For more information, you can also reference the Apache Spark Quick Start Guide.

This first command lists the contents of a folder in the Databricks File System:

Copy

# Take a look at the file system
display(dbutils.fs.ls("/databricks-datasets/samples/docs/"))

The next command uses spark, the SparkSession available in every notebook, to read the README.md text file and create a DataFrame named textFile:

Copy

textFile = spark.read.text("/databricks-datasets/samples/docs/README.md")

To count the lines of the text file, apply the count action to the DataFrame:

Copy

textFile.count()

One thing you may notice is that the second command, reading the text file, does not generate any output while the third command, performing the count, does. The reason for this is that the first command is a transformation while the second one is an action. Transformations are lazy and run only when an action is run. This allows Spark to optimize for performance (for example, run a filter prior to a join), instead of running commands serially. For a complete list of transformations and actions, refer to the Apache Spark Programming Guide: Transformations and Actions.

Apache Spark interfaces

Spark interfaces

There are three key Spark interfaces that you should know about.

Resilient Distributed Dataset (RDD)

Apache Spark’s first abstraction was the RDD. It is an interface to a sequence of data objects that consist of one or more types that are located across a collection of machines (a cluster). RDDs can be created in a variety of ways and are the “lowest level” API available. While this is the original data structure for Apache Spark, you should focus on the DataFrame API, which is a superset of the RDD functionality. The RDD API is available in the Java, Python, and Scala languages.

DataFrame

These are similar in concept to the DataFrame you may be familiar with in the pandas Python library and the R language. The DataFrame API is available in the Java, Python, R, and Scala languages.

 

Dataset

A combination of DataFrame and RDD. It provides the typed interface that is available in RDDs while providing the convenience of the DataFrame. The Dataset API is available in the Java and Scala languages.

In many scenarios, especially with the performance optimizations embedded in DataFrames and Datasets, it will not be necessary to work with RDDs. But it is important to understand the RDD abstraction because:

• The RDD is the underlying infrastructure that allows Spark to run so fast and provide data lineage.
• If you are diving into more advanced components of Spark, it may be necessary to use RDDs.
• The visualizations within the Spark UI reference RDDs.

When you develop Spark applications, you typically use DataFrames and Datasets.

py4j.Py4JException: Method abs([class java.lang.String]) does not exist

# TODO

from pyspark.sql.functions import abs
peopleWithFixedSalariesDF = peopleDF.select(abs("salary")).filter(col("salary")<0);
display(peopleWithFixedSalariesDF)


--------------------------------------------------------------------------- Py4JError Traceback (most recent call last) <command-3332536293827318> in <module>() 2 3 from pyspark.sql.functions import abs ----> 4 peopleWithFixedSalariesDF = peopleDF.select(abs("salary")).filter(col("salary")<0); 5 display(peopleWithFixedSalariesDF) /databricks/spark/python/pyspark/sql/functions.py in _(col) 42 def _(col): 43 sc = SparkContext._active_spark_context ---> 44 jc = getattr(sc._jvm.functions, name)(col._jc if isinstance(col, Column) else col) 45 return Column(jc) 46 _.__name__ = name /databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args) 1255 answer = self.gateway_client.send_command(command) 1256 return_value = get_return_value( -> 1257 answer, self.gateway_client, self.target_id, self.name) 1258 1259 for temp_arg in temp_args: /databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw) 61 def deco(*a, **kw): 62 try: ---> 63 return f(*a, **kw) 64 except py4j.protocol.Py4JJavaError as e: 65 s = e.java_exception.toString() /databricks/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name) 330 raise Py4JError( 331 "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n". --> 332 format(target_id, ".", name, value)) 333 else: 334 raise Py4JError( Py4JError: An error occurred while calling z:org.apache.spark.sql.functions.abs. Trace: py4j.Py4JException: Method abs([class java.lang.String]) does not exist at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:341) at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:362) at py4j.Gateway.invoke(Gateway.java:289) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:251) at java.lang.Thread.run(Thread.java:748)

Solution

Error is due to the fact "salary" is treated as a string and not as a salary column, since it's treated as a string and abs function which accepts string argument doesn't exist thus the error.

To solve use col function to ensure "salary" is treated as a column in DataFrame and not literally as a string.

thus to solve use

from pyspark.sql.functions import abs,col

peopleWithFixedSalariesDF = peopleDF.select(abs(col("salary"))).filter(col("salary")<0);

display(peopleWithFixedSalariesDF)

dataframe : how to groupBy/alias count then filter on count

df.groupBy("x").agg(count("*").alias("cnt"))



top10FemaleFirstNamesDF = (peopleDF.select("firstName").filter("gender=='F'").groupBy("firstName").agg(count("*").alias("cnt")).sort(desc("cnt")).limit(10));

Create a temporary view from Spark DataFrame

Once a temporary view has been created, it can be queried as if it were a table.


peopleDF.createOrReplaceTempView("People10M")

To view the contents of temporary view, use select notation.

display(spark.sql("SELECT * FROM  People10M where firstName = 'Donna' "))

Temporary Views

In DataFrames, temporary views are used to make the DataFrame available to SQL, and work with SQL syntax seamlessly.

A temporary view gives you a name to query from SQL, but unlike a table it exists only for the duration of your Spark Session. As a result, the temporary view will not carry over when you restart the cluster or switch to a new notebook. It also won't show up in the Data button on the menu on the left side of a Databricks notebook which provides easy access to databases and tables.

The statement in the following cells create a temporary view containing the same data.

Visualization - Display Function

Visualization

Databricks provides easy-to-use, built-in visualizations for your data.

Display the data by invoking the Spark display function.

Visualize the query below by selecting the bar graph icon once the table is displayed:

How many women were named Mary in each year?

marysDF = (peopleDF.select(year("birthDate").alias("birthYear")) 

  .filter("firstName = 'Mary' ") 

  .filter("gender = 'F' ") 

  .orderBy("birthYear") 

  .groupBy("birthYear") 

  .count()

)

To start the visualization process, first apply the display function to the DataFrame.

Next, click the graph button in the bottom left corner (second from left) to display data in different ways.

The data initially shows up in html format as an n X 2 column where one column is the birthYear and another column is count.

How to Read a Parquet file into Spark dataframe

peopleDF = spark.read.parquet("/mnt/training/dataframes/people-10m.parquet")


display(peopleDF)