<jupyter_start><jupyter_text>Common Data Wrangling Operation<jupyter_code># initialize spark session
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.dynamicframe import DynamicFrame
from pprint import pprint

# Create SparkContext
sparkContext = SparkContext.getOrCreate()
# Create Glue Context
glueContext = GlueContext(sparkContext)
# Get spark session
spark = glueContext.spark_session
import numpy
import pandas
import pyarrow
import pyspark

package_list = [
    numpy,
    pandas,
    pyarrow,
    pyspark,
]
for package in package_list:
    print(f"{package.__name__}=={package.__version__}")

# import
from datetime import datetime, date, timedelta
    
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pyspark.sql import Row, Column, DataFrame
import pyspark.sql.functions as sql_funcs
import pyspark.sql.types as sql_types<jupyter_output><empty_output><jupyter_text>DataFrame Creation<jupyter_code># 1. create a PySpark DataFrame from a list of rows
pdf = spark.createDataFrame([
    Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),
    Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0)),
    Row(a=4, b=5., c='string3', d=date(2000, 3, 1), e=datetime(2000, 1, 3, 12, 0))
])
pdf
# 2. Create a PySpark DataFrame with an explicit schema.
pdf = spark.createDataFrame([
    (1, 2., 'string1', date(2000, 1, 1), datetime(2000, 1, 1, 12, 0)),
    (2, 3., 'string2', date(2000, 2, 1), datetime(2000, 1, 2, 12, 0)),
    (3, 4., 'string3', date(2000, 3, 1), datetime(2000, 1, 3, 12, 0))
], schema='a long, b double, c string, d date, e timestamp')
pdf
# 3. Create a PySpark DataFrame from a pandas DataFrame
df = pd.DataFrame({
    'a': [1, 2, 3],
    'b': [2., 3., 4.],
    'c': ['string1', 'string2', 'string3'],
    'd': [date(2000, 1, 1), date(2000, 2, 1), date(2000, 3, 1)],
    'e': [datetime(2000, 1, 1, 12, 0), datetime(2000, 1, 2, 12, 0), datetime(2000, 1, 3, 12, 0)]
})
pdf = spark.createDataFrame(df)
pdf
# 4. Create a PySpark DataFrame from an RDD consisting of a list of tuples.
rdd = spark.sparkContext.parallelize([
    (1, 2., 'string1', date(2000, 1, 1), datetime(2000, 1, 1, 12, 0)),
    (2, 3., 'string2', date(2000, 2, 1), datetime(2000, 1, 2, 12, 0)),
    (3, 4., 'string3', date(2000, 3, 1), datetime(2000, 1, 3, 12, 0))
])
pdf = spark.createDataFrame(rdd, schema=['a', 'b', 'c', 'd', 'e'])
pdf
# The DataFrames created above all have the same results and schema.
pdf.printSchema()
pdf.show()<jupyter_output><empty_output><jupyter_text>DataFrame AttributeAccess attribute doesn't trigger any computationRef:- https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.html<jupyter_code># access all columns as list of str
pdf.columns
# access datatype as list of str
pdf.dtypes
# return schema object (not show schema)
pdf.schema<jupyter_output><empty_output><jupyter_text>DataFrame MethodCall method usually triggers computationRef:- https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.html<jupyter_code># count number of rows
pdf.count()
# Return the first ``pyspark.sql.Row``
pdf.first()
# Return first N of ``pyspark.sql.Row`` as a list
pdf.head(3)
# Return first N of ``pyspark.sql.Row`` as a list
pdf.take(3)
# Return last N of ``pyspark.sql.Row`` as a list (added in 3.0.0)
# pdf.tail(3)
# Pretty print first n rows
pdf.show(3, truncate=False, vertical=True)
# access one column method 1
pdf.a
# access one column method 2
pdf["a"]
# access multiple column
pdf[["a", "b"]].show()
# drop column and return new dataframe
pdf.drop("c", "d", "e").show()<jupyter_output><empty_output><jupyter_text>Add Column to DataFrame如果只要添加一列, 用 ``pyspark.sql.DataFrame.withColumn`` 方法. 如果要添加多列, 由于 ``pyspark.sql.DataFrame.withColumn`` 方法需要 projection, 如果用该方法添加多列, 会占用大量内存. 官方建议使用 ``pyspark.sql.DataFrame.select`` 方法同时操作多列.- https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.withColumn.htmlpyspark.sql.DataFrame.withColumn- https://sparkbyexamples.com/pyspark/pyspark-add-new-column-to-dataframe/<jupyter_code># 1. Add New Column with Constant Value
pdf.withColumn("f", sql_funcs.lit("Alice")).show()
# 2. Add Column Based on Another Column of DataFrame
pdf.withColumn("f", pdf.a * pdf.a).show()
# 3. Add Column Value Based on Condition
pdf.withColumn(
    "f", 
    sql_funcs.when(
        pdf.a <= 1, 
        sql_funcs.lit("<=1"),
    ).when(
        (pdf.a > 1) & (pdf.a < 3), 
        sql_funcs.lit("1 < ? < 3"),
    ).otherwise(sql_funcs.lit(">=3"))
).show()
# ``DataFrame.withColumn`` method introduces a projection internally. 
# Therefore, calling it multiple times, for instance, via loops in order to 
# add multiple columns can generate big plans which can cause performance issues 
# and even StackOverflowException. 
# To avoid this, use select() with the multiple columns at once.
pdf.select(
    pdf.a,
    pdf.b,
    pdf.c,
    pdf.d,
    pdf.e,
    (pdf.a * pdf.a).alias("f"),
    (pdf.b * 10).alias("g")
).show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14891990059262092, 'fraction_numerical': 0.046071496845727394, 'mean_word_length': 4.495798319327731, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 23, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '641770', 'license': 'MIT', 'n_tokens_mistral': 2093, 'n_tokens_neox': 1887, 'n_words': 554, 'path': '02-pyspark-examples/04-Data-Wrangling/common-operation.ipynb', 'repo_name': 'MacHu-GWU/learn_big_data_on_aws-project', 'size': '5231'}