Guide - AWS Glue and PySpark 
Tags: aws, cloud, bigdata, pyspark
Published: 2020-09-19 17:25:49

<!-- wp:paragraph -->
<p>In this post, I have penned down AWS Glue and PySpark functionalities which can be helpful when thinking of creating AWS pipeline and writing AWS Glue PySpark scripts.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>AWS Glue is a fully managed extract, transform, and load (ETL) service  to process large amount of datasets from various sources for analytics and data processing.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>While creating the AWS Glue job, you can select between Spark, Spark Streaming and Python shell. These job can run proposed script generated by AWS Glue, or an existing script that you provide or a new script authored by you. Along with this you can select different monitoring options, job execution capacity, timeouts, delayed notification threshold and non-overridable and overridable parameters.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center","id":4165,"width":350,"sizeSlug":"medium"} -->
<div class="wp-block-image"><figure class="aligncenter size-medium is-resized"><img src="https://aprakash.files.wordpress.com/2020/08/screen-shot-2020-08-21-at-6.15.45-pm.png?w=300" alt="" class="wp-image-4165" width="350"/><figcaption>Glue Job Type and Glue Version</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:image {"align":"center","id":4166,"width":350,"sizeSlug":"medium"} -->
<div class="wp-block-image"><figure class="aligncenter size-medium is-resized"><img src="https://aprakash.files.wordpress.com/2020/08/screen-shot-2020-08-21-at-6.16.07-pm.png?w=288" alt="" class="wp-image-4166" width="350"/><figcaption>Script file name and other available options</figcaption></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>AWS recently launched Glue version 2.0 which features 10x faster Spark ETL job start times and reducing the billing duration from a 10 minute minimum to 1 minute minimum.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><a href="https://aws.amazon.com/blogs/aws/aws-glue-version-2-0-featuring-10x-faster-job-start-times-and-1-minute-minimum-billing-duration" target="_blank" rel="noreferrer noopener">https://aws.amazon.com/blogs/aws/aws-glue-version-2-0-featuring-10x-faster-job-start-times-and-1-minute-minimum-billing-duration</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>With AWS Glue you can create development endpoint and configure SageMaker or Zeppelin notebooks to develop and test your Glue ETL scripts. </p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center","id":4172,"width":120,"height":299,"sizeSlug":"medium"} -->
<div class="wp-block-image"><figure class="aligncenter size-medium is-resized"><img src="https://aprakash.files.wordpress.com/2020/08/screen-shot-2020-08-21-at-6.56.12-pm-1.png?w=110" alt="" class="wp-image-4172" width="120" height="299"/></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>I create a SageMaker notebook connected to the Dev endpoint to author and test the ETL scripts. Depending on the language you are comfortable with, you can spin up the notebook.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"align":"center","id":4181,"sizeSlug":"large"} -->
<div class="wp-block-image"><figure class="aligncenter size-large"><img src="https://aprakash.files.wordpress.com/2020/08/screen-shot-2020-08-21-at-7.06.00-pm-1.png?w=1024" alt="" class="wp-image-4181"/></figure></div>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>Now, lets talk about some specific features and functionalities in AWS Glue and PySpark which can be helpful.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>1. <strong><span style="text-decoration:underline;">Spark DataFrames</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Spark DataFrame is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database. You can create DataFrame from RDD, from file formats like csv, json, parquet.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>With SageMaker Sparkmagic(PySpark) Kernel notebook, Spark session is automatically created.</p>
<!-- /wp:paragraph -->

<!-- wp:image {"id":4104,"sizeSlug":"large","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><img src="https://aprakash.files.wordpress.com/2020/07/screen-shot-2020-07-04-at-9.43.19-pm.png?w=699" alt="" class="wp-image-4104"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>To create DataFrame -</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code># from CSV files 
S3_IN = "s3://mybucket/train/training.csv"

csv_df = (
    spark.read.format("org.apache.spark.csv")
    .option("header", True)
    .option("quote", '"')
    .option("escape", '"')
    .option("inferSchema", True)
    .option("ignoreLeadingWhiteSpace", True)
    .option("ignoreTrailingWhiteSpace", True)
    .csv(S3_IN, multiLine=False)
)

# from PARQUET files 
S3_PARQUET="s3://mybucket/folder1/dt=2020-08-24-19-28/"

df = spark.read.parquet(S3_PARQUET)

# from JSON files
df = spark.read.json(S3_JSON)

# from multiline JSON file 
df = spark.read.json(S3_JSON, multiLine=True)</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>2. <strong><span style="text-decoration:underline;">GlueContext</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>GlueContext is the entry point for reading and writing DynamicFrames in AWS Glue. It wraps the Apache SparkSQL SQLContext object providing mechanisms for interacting with the Apache Spark platform.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>from awsglue.job import Job
from awsglue.transforms import *
from awsglue.context import GlueContext
from pyspark.context import SparkContext
from awsglue.utils import getResolvedOptions
from awsglue.dynamicframe import DynamicFrame

glueContext = GlueContext(SparkContext.getOrCreate())</code></pre>
<!-- /wp:code -->

<!-- wp:image {"id":4185,"sizeSlug":"large","className":"is-style-default"} -->
<figure class="wp-block-image size-large is-style-default"><img src="https://aprakash.files.wordpress.com/2020/08/screen-shot-2020-08-21-at-8.03.09-pm.png?w=1024" alt="" class="wp-image-4185"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>3. <strong><span style="text-decoration:underline;">DynamicFrame</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>AWS Glue DynamicFrames are similar to SparkSQL DataFrames. It represent a distributed collection of data without requiring you to specify a schema.It can also be used to read and transform data that contains inconsistent values and types. </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>DynamicFrame can be created using the below options –</p>
<!-- /wp:paragraph -->

<!-- wp:list -->
<ul><li><em>create_dynamic_frame_from_rdd</em> – created from an Apache Spark Resilient Distributed Dataset (RDD)</li><li><em>create_dynamic_frame_from_catalog</em> – created using a Glue catalog database and table name</li><li><em>create_dynamic_frame_from_options</em> – created with the specified connection and format. Example – The connection type, such as Amazon S3, Amazon Redshift, and JDBC</li></ul>
<!-- /wp:list -->

<!-- wp:paragraph -->
<p>DynamicFrames can be converted to and from DataFrames using .toDF() and fromDF().</p>
<!-- /wp:paragraph -->

<!-- wp:code {"ampFitText":true} -->
<amp-fit-text layout="fixed-height" min-font-size="6" max-font-size="10" height="80"><pre class="wp-block-code"><code>#create DynamicFame from S3 parquet files
datasource0 = glueContext.create_dynamic_frame_from_options(
            connection_type="s3",
            connection_options = {
                "paths": &#91;S3_location]
            },
            format="parquet",
            transformation_ctx="datasource0")

#create DynamicFame from glue catalog 
datasource0 = glueContext.create_dynamic_frame.from_catalog(
           database = "demo",
           table_name = "testtable",
           transformation_ctx = "datasource0")

#convert to spark DataFrame 
df1 = datasource0.toDF()

#convert to Glue DynamicFrame
df2 = DynamicFrame.fromDF(df1, glueContext , "df2")
</code></pre></amp-fit-text>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>Further Read - <a rel="noreferrer noopener" href="https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_catalog" target="_blank">https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_catalog</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>4. <strong><span style="text-decoration:underline;">AWS Glue Job Bookmark </span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>AWS Glue<em> Job bookmark</em> helps process incremental data when rerunning the job on a scheduled interval, preventing reprocessing of old data.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further Read  - <a href="https://aprakash.wordpress.com/2020/05/07/implementing-glue-etl-job-with-job-bookmarks/">https://aprakash.wordpress.com/2020/05/07/implementing-glue-etl-job-with-job-bookmarks/</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><a href="https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-glue-arguments.html">https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-glue-arguments.html</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>5. <strong><span style="text-decoration:underline;">Write out data</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>The DynamicFrame of transformed dataset can be written out to S3 as non-partitioned (default) or partitioned. "<em>partitionKeys</em>" parameter can be specified in connection_option to write out the data to S3 as partitioned. AWS Glue organizes these dataset in Hive-style partition.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>In the below code example, AWS Glue DynamicFrame is partitioned by year, month, day, hour and written in parquet format in Hive-style partition on to S3.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>s3://bucket_name/table_name/year=2020/month=7/day=13/hour=14/part-000-671c.c000.snappy.parquet</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>S3_location = "s3://bucket_name/table_name"

datasink = glueContext.write_dynamic_frame_from_options(
    frame= data,
    connection_type="s3",
    connection_options={
        "path": S3_location,
        "partitionKeys": &#91;"year", "month", "day", "hour"]
    },
    format="parquet",
    transformation_ctx ="datasink")</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>Further Read - <a href="https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options">https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>6. <strong><span style="text-decoration:underline;">"glueparquet" format option</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>glueparquet is a performance optimized Apache parquet writer type for writing DynamicFrames. It computes and modifies the schema dynamically. </p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>datasink = glueContext.write_dynamic_frame_from_options(
               frame=dynamicframe,
               connection_type="s3",
               connection_options={
                  "path": S3_location,
                  "partitionKeys": &#91;"year", "month", "day", "hour"]
               },
               format="glueparquet",
               format_options = {"compression": "snappy"},
               transformation_ctx ="datasink")</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>Further Read - <a href="https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-format.html" target="_blank" rel="noreferrer noopener">https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-format.html</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>7. <strong><span style="text-decoration:underline;">S3 Lister and other options for optimizing memory management </span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>AWS Glue provides an optimized mechanism to list files on S3 while reading data into DynamicFrame which can be enabled using additional_options parameter "<em>useS3ListImplementation</em>" to true.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further Read - <a href="https://aws.amazon.com/blogs/big-data/optimize-memory-management-in-aws-glue/">https://aws.amazon.com/blogs/big-data/optimize-memory-management-in-aws-glue/</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>8. <strong><span style="text-decoration:underline;">Purge S3 path</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><em>purge_s3_path</em> is a nice option available to delete files from specified S3 path recursively based on retention period or other available filters. As an example, suppose you are running AWS Glue job to fully refresh the table per day writing the data to S3 with naming convention of <em>s3://bucket-name/table-name/dt=&lt;data-time&gt;.</em>  Based on the defined retention period using the Glue job itself you can delete the dt=&lt;date-time&gt; s3 folders. Another option is to set S3 bucket lifecycle policy with prefix.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>#purge locations older than 3 days
print("Attempting to purge S3 path with retention set to 3 days.")
glueContext.purge_s3_path(
    s3_path=output_loc, 
    options={"retentionPeriod": 72})</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>You have other options like purge_table, transition_table and transition_s3_path also available. The transition_table option transitions the storage class of the files stored on Amazon S3 for the specified catalog's database and table.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further Read - <a href="https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-purge_s3_path">https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html#aws-glue-api-crawler-pyspark-extensions-glue-context-purge_s3_path</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>9. <strong><span style="text-decoration:underline;">Relationalize Class</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><em>Relationalize</em> class can help flatten nested json outermost level. </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further read - <a href="https://aprakash.wordpress.com/2020/02/26/aws-glue-querying-nested-json-with-relationalize-transform/">https://aprakash.wordpress.com/2020/02/26/aws-glue-querying-nested-json-with-relationalize-transform/</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>10. <strong><span style="text-decoration:underline;">Unbox Class</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>The <em>Unbox</em> class helps unbox string field in DynamicFrame to specified format type(optional).</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Further read - <a href="https://aprakash.wordpress.com/2020/02/26/aws-glue-querying-nested-json-with-relationalize-transform/">https://aprakash.wordpress.com/2020/02/26/aws-glue-querying-nested-json-with-relationalize-transform/</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>11. <strong><span style="text-decoration:underline;">Unnest Class </span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>The <em>Unnest</em> class flattens nested objects to top-level elements in a DynamicFrame.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">root
|-- id: string
|-- type: string
|-- content: map
|    |-- keyType: string
|    |-- valueType: string</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>With content attribute/column being map Type, we can use unnest class to unnest each key elements.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>unnested = UnnestFrame.apply(frame=data_dynamic_dframe)
unnested.printSchema()</code></pre>
<!-- /wp:code -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">root
|-- id: string
|-- type: string
|-- content.dateLastUpdated: string
|-- content.creator: string
|-- content.dateCreated: string
|-- content.title: string</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>12. <strong><span style="text-decoration:underline;">printSchema()</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>To print the Spark or Glue DynamicFrame schema in tree format use <em>printSchema()</em>.</p>
<!-- /wp:paragraph -->

<!-- wp:preformatted -->
<pre class="wp-block-preformatted">datasource0.printSchema()

root
|-- ID: int
|-- Name: string
|-- Identity: string
|-- Alignment: string
|-- EyeColor: string
|-- HairColor: string
|-- Gender: string
|-- Status: string
|-- Appearances: int
|-- FirstAppearance: choice
|    |-- int
|    |-- long
|    |-- string
|-- Year: int
|-- Universe: string
</pre>
<!-- /wp:preformatted -->

<!-- wp:paragraph -->
<p>13. <strong><span style="text-decoration:underline;">Fields Selection</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><em>select_fields</em> can be used to select fields from Glue DynamicFrame.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code># From DynamicFrame

datasource0.select_fields(&#91;"Status","HairColor"]).toDF().distinct().show()</code></pre>
<!-- /wp:code -->

<!-- wp:image {"id":4110,"sizeSlug":"large"} -->
<figure class="wp-block-image size-large"><img src="https://aprakash.files.wordpress.com/2020/07/screen-shot-2020-07-05-at-12.14.07-am.png?w=700" alt="" class="wp-image-4110"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>To select fields from Spark Dataframe use "<em>select</em>" - </p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code># From Dataframe

datasource0_df.select(&#91;"Status","HairColor"]).distinct().show()</code></pre>
<!-- /wp:code -->

<!-- wp:image {"id":4112,"sizeSlug":"large"} -->
<figure class="wp-block-image size-large"><img src="https://aprakash.files.wordpress.com/2020/07/screen-shot-2020-07-05-at-12.16.33-am.png?w=661" alt="" class="wp-image-4112"/></figure>
<!-- /wp:image -->

<!-- wp:paragraph -->
<p>14. <strong><span style="text-decoration:underline;">Timestamp</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Suppose the application writes data into DynamoDB and has last_updated attribute/column.  DynamoDB does not natively support date/timestamp data type. So, you could either store it as String or Number. If stored as number, its usually done as epoch time - the number of seconds since 00:00:00 UTC on 1 January 1970. You could see something like "1598331963" which is 2020-08-25T05:06:03+00:00 in ISO 8601.</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p><a rel="noreferrer noopener" href="https://www.unixtimestamp.com/index.php" target="_blank">https://www.unixtimestamp.com/index.php</a></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>How can you convert it to timestamp?</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>When you read the data using AWS Glue DynamicFrame and view the schema, it will show it as "long" data type.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>root
|-- version: string
|-- item_id: string
|-- status: string
|-- event_type: string
|-- last_updated: long</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>To convert the last_updated long data type into timestamp data type, you can use the below -</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>import pyspark.sql.functions as f
import pyspark.sql.types as t

new_df = (
    df
        .withColumn("last_updated", f.from_unixtime(f.col("last_updated")/1000).cast(t.TimestampType()))
)   </code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>15. <strong><span style="text-decoration:underline;">Temporary View from Spark DataFrame</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>If you want to store the Spark DataFrame as table and query it using spark sql, you can convert the DataFrame into temporary view that is available for only that spark session using  <em>createOrReplaceTempView</em>.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>df = spark.createDataFrame(
    &#91;
        (1, &#91;'a', 'b', 'c'], 90.00),
        (2, &#91;'x', 'y'], 99.99),
    ],
    &#91;'id', 'event', 'score'] 
)

df.printSchema()
root
 |-- id: long (nullable = true)
 |-- event: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- score: double (nullable = true)

df.createOrReplaceTempView("example")

spark.sql("select * from example").show()

+---+---------+-----+
| id|    event|score|
+---+---------+-----+
|  1|&#91;a, b, c]| 90.0|
|  2|   &#91;x, y]|99.99|
+---+---------+-----+</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>16. <strong><span style="text-decoration:underline;">Extract element from ArrayType</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Suppose from the above example, you want to create a new attribute/column to store only the last event. How would you do it? </p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>Using <em>element_at</em> function. It returns element of array at given index in extraction if col is array. It can also be used to extract given key in extraction if col is map.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>import pyspark.sql.functions as element_at

newdf = df.withColumn("last_event", element_at("event", -1))

newdf.printSchema()
root
 |-- id: long (nullable = true)
 |-- event: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- score: double (nullable = true)
 |-- last_event: string (nullable = true)

newdf.show()
+---+---------+-----+----------+
| id|    event|score|last_event|
+---+---------+-----+----------+
|  1|&#91;a, b, c]| 90.0|         c|
|  2|   &#91;x, y]|99.99|         y|
+---+---------+-----+----------+</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>17. <strong><span style="text-decoration:underline;">explode</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>The <em>explode</em> function in PySpark is used to explode array or map columns in rows. Taking an example, lets try to explode "event" column from the above example</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>from pyspark.sql.functions import explode

df1 = df.select(df.id,explode(df.event))

df1.printSchema()
root
 |-- id: long (nullable = true)
 |-- col: string (nullable = true)

df1.show()
+---+---+
| id|col|
+---+---+
|  1|  a|
|  1|  b|
|  1|  c|
|  2|  x|
|  2|  y|
+---+---+</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p> 18. <strong><span style="text-decoration:underline;">getField</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>In a Struct type, if you want to get a field by name, you can use "<em>getField</em>".</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>import pyspark.sql.functions as f
from pyspark.sql import Row

from pyspark.sql import Row
df = spark.createDataFrame(&#91;Row(attributes=Row(Name='scott', Height=6.0, Hair='black')),
                            Row(attributes=Row(Name='kevin', Height=6.1, Hair='brown'))]
)

df.printSchema()
root
 |-- attributes: struct (nullable = true)
 |    |-- Hair: string (nullable = true)
 |    |-- Height: double (nullable = true)
 |    |-- Name: string (nullable = true)

df.show()
+-------------------+
|         attributes|
+-------------------+
|&#91;black, 6.0, scott]|
|&#91;brown, 6.1, kevin]|
+-------------------+

df1 = (df
      .withColumn("name", f.col("attributes").getField("Name"))
      .withColumn("height", f.col("attributes").getField("Height"))
      .drop("attributes")
      )

df1.show()
+-----+------+
| name|height|
+-----+------+
|scott|   6.0|
|kevin|   5.1|
+-----+------+</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>19. <strong><span style="text-decoration:underline;">startswith</span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>If you want to find records based on string match you can use "<em>startswith</em>".</p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>In the below example I am searching for all records where value for description column starts with "[{".</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>import pyspark.sql.functions as f

df.filter(f.col("description").startswith("&#91;{")).show()</code></pre>
<!-- /wp:code -->

<!-- wp:paragraph -->
<p>20. <strong><span style="text-decoration:underline;">Extract year,  month,  day,  hour </span></strong></p>
<!-- /wp:paragraph -->

<!-- wp:paragraph -->
<p>One of the common use case is to write the AWS Glue DynamicFrame or Spark DataFrame to S3 in Hive-style partition. To do so you can extract year, month, day, hour and use it as partitionkeys to write the DynamicFrame/DataFrame to S3.</p>
<!-- /wp:paragraph -->

<!-- wp:code -->
<pre class="wp-block-code"><code>import pyspark.sql.functions as f

df2 = (raw_df
        .withColumn('year', f.year(f.col('last_updated')))
        .withColumn('month', f.month(f.col('last_updated')))
        .withColumn('day', f.dayofmonth(f.col('last_updated')))
        .withColumn('hour', f.hour(f.col('last_updated')))            
        )</code></pre>
<!-- /wp:code -->