Working with Spark 3/Java 11/Python 3.8

Author: dolsysmith

Dockerfile-loader

@@ -1,4 +1,4 @@
-FROM python:3.6-stretch
+FROM python:3.8-buster
 MAINTAINER TweetSets <[email protected]>
 
 ADD requirements.txt /opt/tweetsets/
@@ -7,7 +7,7 @@ RUN pip install -r requirements.txt
 RUN grep elasticsearch-dsl requirements.txt | xargs pip install -t dependencies
 
 RUN apt-get update && \
-    apt-get install -y openjdk-8-jre-headless \
+    apt-get install -y openjdk-11-jre-headless \
     ca-certificates-java \
     zip -y
 
@@ -22,7 +22,7 @@ ADD spark_utils.py /opt/tweetsets/
 ADD tweetsets_schema.json /opt/tweetsets/
 ADD tweetsets_sql_exp.sql /opt/tweetsets
 ADD setup.py /opt/tweetsets/
-ADD elasticsearch-hadoop-7.9.2.jar /opt/tweetsets/elasticsearch-hadoop.jar
+ADD elasticsearch-spark-30_2.12-7.13.4.jar /opt/tweetsets/elasticsearch-hadoop.jar
 ADD tweetset_cli.py /opt/tweetsets/
 
 RUN python setup.py bdist_egg

Dockerfile-spark

@@ -1,4 +1,4 @@
-FROM python:3.6-stretch
+FROM python:3.8-buster
 MAINTAINER TweetSets <[email protected]> 
 
 # Based on https://hub.docker.com/r/gettyimages/spark/~/dockerfile/
@@ -24,15 +24,14 @@ ENV PYTHONIOENCODING UTF-8
 ENV PIP_DISABLE_PIP_VERSION_CHECK 1
 
 # JAVA
-ARG JAVA_MAJOR_VERSION=8
-ARG JAVA_UPDATE_VERSION=131
+ARG JAVA_MAJOR_VERSION=11
+ARG JAVA_UPDATE_VERSION=11+9
 ARG JAVA_BUILD_NUMBER=11
-ENV JAVA_HOME /usr/jdk1.${JAVA_MAJOR_VERSION}.0_${JAVA_UPDATE_VERSION}
+ENV JAVA_HOME /usr/jdk-${JAVA_MAJOR_VERSION}.0.${JAVA_UPDATE_VERSION}
 
 ENV PATH $PATH:$JAVA_HOME/bin
 RUN curl -sL --retry 3 --insecure \
-  --header "Cookie: oraclelicense=accept-securebackup-cookie;" \
-  "http://download.oracle.com/otn-pub/java/jdk/${JAVA_MAJOR_VERSION}u${JAVA_UPDATE_VERSION}-b${JAVA_BUILD_NUMBER}/d54c1d3a095b4ff2b6607d096fa80163/server-jre-${JAVA_MAJOR_VERSION}u${JAVA_UPDATE_VERSION}-linux-x64.tar.gz" \
+  "https://github.com/AdoptOpenJDK/openjdk11-binaries/releases/download/jdk-11.0.11+9/OpenJDK11U-jdk_x64_linux_hotspot_11.0.11_9.tar.gz" \
   | gunzip \
   | tar x -C /usr/ \
   && ln -s $JAVA_HOME /usr/java \
@@ -51,13 +50,13 @@ RUN curl -sL --retry 3 \
  && chown -R root:root $HADOOP_HOME
 
 # SPARK
-ENV SPARK_VERSION 2.4.7
+ENV SPARK_VERSION 3.1.2
 ENV SPARK_PACKAGE spark-${SPARK_VERSION}-bin-without-hadoop
 ENV SPARK_HOME /usr/spark-${SPARK_VERSION}
 ENV SPARK_DIST_CLASSPATH="$HADOOP_HOME/etc/hadoop/*:$HADOOP_HOME/share/hadoop/common/lib/*:$HADOOP_HOME/share/hadoop/common/*:$HADOOP_HOME/share/hadoop/hdfs/*:$HADOOP_HOME/share/hadoop/hdfs/lib/*:$HADOOP_HOME/share/hadoop/hdfs/*:$HADOOP_HOME/share/hadoop/yarn/lib/*:$HADOOP_HOME/share/hadoop/yarn/*:$HADOOP_HOME/share/hadoop/mapreduce/lib/*:$HADOOP_HOME/share/hadoop/mapreduce/*:$HADOOP_HOME/share/hadoop/tools/lib/*"
 ENV PATH $PATH:${SPARK_HOME}/bin
 RUN curl -sL --retry 3 \
-  "https://archive.apache.org/dist/spark/spark-2.4.7/spark-2.4.7-bin-without-hadoop.tgz" \
+  "https://mirrors.sonic.net/apache/spark/spark-3.1.2/spark-3.1.2-bin-without-hadoop.tgz" \
   | gunzip \
   | tar x -C /usr/ \
  && mv /usr/$SPARK_PACKAGE $SPARK_HOME \

elasticsearch-spark-30_2.12-7.13.4.jar

@@ -25,4 +25,4 @@ six==1.10.0
 vine==1.3.0
 Werkzeug==1.0.1
 twarc==1.12.1
-pyspark==2.4.7
+pyspark==3.1.2

requirements.txt

@@ -16,7 +16,8 @@ def make_spark_df(spark, schema, sql, path_to_dataset, dataset_id):
     # Read JSON files as Spark DataFrame
     df = spark.read.schema(schema).json(path_to_dataset)
     # Add the full Tweet JSON as a separate field
-    df = df.withColumn("tweet", F.to_json(F.struct([df[x] for x in df.columns])))
+    # Option for Spark v3 to write null fields as nulls (not skip)
+    df = df.withColumn("tweet", F.to_json(F.struct([df[x] for x in df.columns]), {'ignoreNullFields': 'false'}))
     df.createOrReplaceTempView("tweets")
     # Apply SQL transform
     df = spark.sql(sql)

spark_utils.py

@@ -312,6 +312,11 @@ def shard_count(tweet_count, store_tweet=True):
         tweet_index.create()
 
         spark = SparkSession.builder.appName('TweetSets').getOrCreate()
+        # Make Spark v3 use the v2 time parser
+        # TO DO --> update Spark SQL code to use the new time parser
+        spark.conf.set("spark.sql.legacy.timeParserPolicy","LEGACY")
+        # Set UTC as the time zone
+        spark.conf.set('spark.sql.session.timeZone', 'UTC')
         try:
             es_conf = {"es.nodes": os.environ.get('ES_HOST', 'elasticsearch'),
                        "es.port": "9200",