Introducing generative AI upgrades for Apache Spark in AWS Glue (preview)

Organizations run tens of millions of Apache Spark functions every month on AWS, transferring, processing, and making ready knowledge for analytics and machine studying. As these functions age, holding them safe and environment friendly turns into more and more difficult. Information practitioners must improve to the newest Spark releases to learn from efficiency enhancements, new options, bug fixes, and safety enhancements. Nonetheless, these upgrades are sometimes complicated, expensive, and time-consuming.

As we speak, we’re excited to announce the preview of generative AI upgrades for Spark, a brand new functionality that permits knowledge practitioners to shortly improve and modernize their Spark functions working on AWS. Beginning with Spark jobs in AWS Glue, this characteristic lets you improve from an older AWS Glue model to AWS Glue model 4.0. This new functionality reduces the time knowledge engineers spend on modernizing their Spark functions, permitting them to deal with constructing new knowledge pipelines and getting worthwhile analytics quicker.

Understanding the Spark improve problem

The standard means of upgrading Spark functions requires important guide effort and experience. Information practitioners should rigorously evaluate incremental Spark launch notes to grasp the intricacies and nuances of breaking modifications, a few of which can be undocumented. They then want to switch their Spark scripts and configurations, updating options, connectors, and library dependencies as wanted.

Testing these upgrades includes working the appliance and addressing points as they come up. Every take a look at run could reveal new issues, leading to a number of iterations of modifications. After the upgraded utility runs efficiently, practitioners should validate the brand new output in opposition to the anticipated leads to manufacturing. This course of typically turns into year-long initiatives that price tens of millions of {dollars} and eat tens of hundreds of engineering hours.

How generative AI upgrades for Spark works

The Spark upgrades characteristic makes use of AI to automate each the identification and validation of required modifications to your AWS Glue Spark functions. Let’s discover how these capabilities work collectively to simplify your improve course of.

AI-driven improve plan technology

While you provoke an improve, the service analyzes your utility utilizing AI to establish vital modifications throughout each PySpark code and Spark configurations. Throughout preview, Spark Upgrades helps upgrading from Glue 2.0 (Spark 2.4.3, Python 3.7) to Glue 4.0 (Spark 3.3.0, Python 3.10), routinely dealing with modifications that may usually require in depth guide evaluate of public Spark, Python and Glue model migration guides, adopted by improvement, testing, and verification. Spark Upgrades addresses 4 key areas of modifications:

• Spark SQL API strategies and features
• Spark DataFrame API strategies and operations
• Python language updates (together with module deprecations and syntax modifications)
• Spark SQL and Core configuration settings

The complexity of those upgrades turns into evident when you think about migrating from Spark 2.4.3 to Spark 3.3.0 includes over 100 version-specific modifications. A number of components contribute to the challenges of performing guide upgrades:

• Extremely expressive language with a mixture of crucial and declarative programming kinds, permits customers to simply develop Spark functions. Nonetheless, this will increase the complexity of figuring out impacted code throughout upgrades.
• Lazy execution of transformations in a distributed Spark utility improves efficiency however makes runtime verification of utility upgrades difficult for customers.
• Spark configurations modifications in default values or the introduction of recent configurations throughout variations can influence utility habits in numerous methods, making it troublesome for customers to establish points throughout upgrades.

For instance, in Spark 3.2, Spark SQL TRANSFORM operator can’t assist alias in inputs. In Spark 3.1 and earlier, you can write a script remodel like SELECT TRANSFORM(a AS c1, b AS c2) USING 'cat' FROM TBL.

# Unique code (Glue 2.0)
question = """
SELECT TRANSFORM(merchandise as product_name, value as product_price, quantity as product_number)
   USING 'cat'
FROM items
WHERE items.value > 5
"""
spark.sql(question)

# Up to date code (Glue 4.0)
question = """
SELECT TRANSFORM(merchandise, value, quantity)
   USING 'cat' AS (product_name, product_price, product_number)
FROM items
WHERE items.value > 5
"""
spark.sql(question)

In Spark 3.1, loading and saving timestamps earlier than 1900-01-01 00:00:00Z as INT96 in Parquet information causes errors. In Spark 3.0, this wouldn’t fail however might end in timestamp shifts on account of calendar rebasing. To revive the outdated habits in Spark 3.1, you would wish to configure the Spark SQL configurations for spark.sql.legacy.parquet.int96RebaseModeInRead and spark.sql.legacy.parquet.int96RebaseModeInWrite to LEGACY.

# Unique code (Glue 2.0)
knowledge = [(1, "1899-12-31 23:59:59"), (2, "1900-01-01 00:00:00")]
schema = StructType([ StructField("id", IntegerType(), True), StructField("timestamp", TimestampType(), True) ])
df = spark.createDataFrame(knowledge, schema=schema)
df.write.mode("overwrite").parquet("path/to/parquet_file") 

# Up to date code (Glue 4.0)
qspark.conf.set("spark.sql.legacy.parquet.int96RebaseModeInRead", "LEGACY") 
spark.conf.set("spark.sql.legacy.parquet.int96RebaseModeInWrite", "LEGACY")

knowledge = [(1, "1899-12-31 23:59:59"), (2, "1900-01-01 00:00:00")]
schema = StructType([ StructField("id", IntegerType(), True), StructField("timestamp", TimestampType(), True) ])
df = spark.createDataFrame(knowledge, schema=schema)
df.write.mode("overwrite").parquet("path/to/parquet_file")

Automated validation in your setting

After figuring out the mandatory modifications, Spark Upgrades validates the upgraded utility by working it as an AWS Glue job in your AWS account. The service iterates by means of a number of validation runs, as much as 10, reviewing any errors encountered in every iteration and refining the improve plan till it achieves a profitable run. You possibly can run a Spark Improve Evaluation in your improvement account utilizing mock datasets equipped by means of Glue job parameters used for validation runs.

After Spark Upgrades has efficiently validated the modifications, it presents an improve plan so that you can evaluate. You possibly can then settle for and apply the modifications to your job within the improvement account, earlier than replicating them to your job within the manufacturing account. The Spark Improve plan contains the next:

• An improve abstract with an evidence of code updates made through the course of
• The ultimate script that you should utilize rather than your present script
• Logs from validation runs displaying how points had been recognized and resolved

You possibly can evaluate all features of the improve, together with intermediate validation makes an attempt and any error resolutions, earlier than deciding to use the modifications to your manufacturing job. This method ensures you may have full visibility into and management over the improve course of whereas benefiting from AI-driven automation.

Get began with generative AI Spark upgrades

Let’s stroll by means of the method of upgrading an AWS Glue 2.0 job to AWS Glue 4.0. Full the next steps:

1. On the AWS Glue console, select ETL jobs within the navigation pane.
2. Choose your AWS Glue 2.0 job, and select Run improve evaluation with AI.
   
3. For Outcome path, enter s3://aws-glue-assets-<account-id>-<area>/scripts/upgraded/ (present your individual account ID and AWS Area).
4. Select Run.
   
5. On the Improve evaluation tab, watch for the evaluation to be accomplished.
   
   Whereas an evaluation is working, you’ll be able to view the intermediate job evaluation makes an attempt (as much as 10) for validation below the Runs tab. Moreover, the Upgraded abstract in S3 paperwork the upgrades made by the Spark Improve service up to now, refining the improve plan with every try. Every try will show a distinct failure cause, which the service tries to deal with within the subsequent try by means of code or configuration updates.
   After a profitable evaluation, the upgraded script and a abstract of modifications can be uploaded to Amazon Easy Storage Service (Amazon S3).
6. Assessment the modifications to verify they meet your necessities, then select Apply upgraded script.
   

Your job has now been efficiently upgraded to AWS Glue model 4.0. You possibly can test the Script tab to confirm the up to date script and the Job particulars tab to evaluate the modified configuration.

Understanding the improve course of by means of an instance

We now present a manufacturing Glue 2.0 job that we want to improve to Glue 4.0 utilizing the Spark Improve characteristic. This Glue 2.0 job reads a dataset, up to date each day in an S3 bucket below totally different partitions, containing new guide opinions from a web-based market and runs SparkSQL to collect insights into the consumer votes for the guide opinions.

Unique code (Glue 2.0) – earlier than improve

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
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
from collections import Sequence
from pyspark.sql.sorts import DecimalType
from pyspark.sql.features import lit, to_timestamp, col

def is_data_type_sequence(coming_dict):
    return True if isinstance(coming_dict, Sequence) else False

def dataframe_to_dict_list(df):
    return [row.asDict() for row in df.collect()]

books_input_path = (
    "s3://aws-bigdata-blog/generated_synthetic_reviews/knowledge/product_category=Books/"
)
view_name = "books_temp_view"
static_date = "2010-01-01"
books_source_df = (
    spark.learn.choice("header", "true")
    .choice("recursiveFileLookup", "true")
    .choice("path", books_input_path)
    .parquet(books_input_path)
)
books_source_df.createOrReplaceTempView(view_name)
books_with_new_review_dates_df = spark.sql(
    f"""
        SELECT 
        {view_name}.*,
            DATE_ADD(to_date(review_date), "180.8") AS next_review_date,
            CASE 
                WHEN DATE_ADD(to_date(review_date), "365") < to_date('{static_date}') THEN 'Sure' 
                ELSE 'No' 
            END AS Actionable
        FROM {view_name}
    """
)
books_with_new_review_dates_df.createOrReplaceTempView(view_name)
aggregate_books_by_marketplace_df = spark.sql(
    f"SELECT market, depend({view_name}.*) as total_count, avg(star_rating) as average_star_ratings, avg(helpful_votes) as average_helpful_votes, avg(total_votes) as average_total_votes  FROM {view_name} group by market"
)
aggregate_books_by_marketplace_df.present()
knowledge = dataframe_to_dict_list(aggregate_books_by_marketplace_df)
if is_data_type_sequence(knowledge):
    print("knowledge is legitimate")
else:
    elevate ValueError("Information is invalid")

aggregated_target_books_df = aggregate_books_by_marketplace_df.withColumn(
    "average_total_votes_decimal", col("average_total_votes").forged(DecimalType(3, -2))
)
aggregated_target_books_df.present()

New code (Glue 4.0) – after improve

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 collections.abc import Sequence
from pyspark.sql.sorts import DecimalType
from pyspark.sql.features import lit, to_timestamp, col

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.legacy.allowStarWithSingleTableIdentifierInCount", "true")
spark.conf.set("spark.sql.legacy.allowNegativeScaleOfDecimal", "true")
job = Job(glueContext)

def is_data_type_sequence(coming_dict):
    return True if isinstance(coming_dict, Sequence) else False

def dataframe_to_dict_list(df):
    return [row.asDict() for row in df.collect()]

books_input_path = (
    "s3://aws-bigdata-blog/generated_synthetic_reviews/knowledge/product_category=Books/"
)
view_name = "books_temp_view"
static_date = "2010-01-01"
books_source_df = (
    spark.learn.choice("header", "true")
    .choice("recursiveFileLookup", "true")
    .load(books_input_path)
)
books_source_df.createOrReplaceTempView(view_name)
books_with_new_review_dates_df = spark.sql(
    f"""
        SELECT 
        {view_name}.*,
            DATE_ADD(to_date(review_date), 180) AS next_review_date,
            CASE 
                WHEN DATE_ADD(to_date(review_date), 365) < to_date('{static_date}') THEN 'Sure' 
                ELSE 'No' 
            END AS Actionable
        FROM {view_name}
    """
)
books_with_new_review_dates_df.createOrReplaceTempView(view_name)
aggregate_books_by_marketplace_df = spark.sql(
    f"SELECT market, depend({view_name}.*) as total_count, avg(star_rating) as average_star_ratings, avg(helpful_votes) as average_helpful_votes, avg(total_votes) as average_total_votes  FROM {view_name} group by market"
)
aggregate_books_by_marketplace_df.present()
knowledge = dataframe_to_dict_list(aggregate_books_by_marketplace_df)
if is_data_type_sequence(knowledge):
    print("knowledge is legitimate")
else:
    elevate ValueError("Information is invalid")

aggregated_target_books_df = aggregate_books_by_marketplace_df.withColumn(
    "average_total_votes_decimal", col("average_total_votes").forged(DecimalType(3, -2))
)
aggregated_target_books_df.present()

Improve abstract

In Spark 3.2, spark.sql.adaptive.enabled is enabled by default. To revive the habits earlier than Spark 3.2, 
you'll be able to set spark.sql.adaptive.enabled to false.

No appropriate migration rule was discovered within the supplied context for this particular error. The change was made based mostly on the error message, which indicated that Sequence couldn't be imported from collections module. In Python 3.10, Sequence has been moved to the collections.abc module.

In Spark 3.1, path choice can't coexist when the next strategies are known as with path parameter(s): DataFrameReader.load(), DataFrameWriter.save(), DataStreamReader.load(), or DataStreamWriter.begin(). As well as, paths choice can't coexist for DataFrameReader.load(). For instance, spark.learn.format(csv).choice(path, /tmp).load(/tmp2) or spark.learn.choice(path, /tmp).csv(/tmp2) will throw org.apache.spark.sql.AnalysisException. In Spark model 3.0 and under, path choice is overwritten if one path parameter is handed to above strategies; path choice is added to the general paths if a number of path parameters are handed to DataFrameReader.load(). To revive the habits earlier than Spark 3.1, you'll be able to set spark.sql.legacy.pathOptionBehavior.enabled to true.

In Spark 3.0, the `date_add` and `date_sub` features accepts solely int, smallint, tinyint because the 2nd argument; fractional and non-literal strings usually are not legitimate anymore, for instance: `date_add(forged('1964-05-23' as date), '12.34')` causes `AnalysisException`. Notice that, string literals are nonetheless allowed, however Spark will throw `AnalysisException` if the string content material will not be a sound integer. In Spark model 2.4 and under, if the 2nd argument is fractional or string worth, it's coerced to int worth, and the result's a date worth of `1964-06-04`.

In Spark 3.2, the utilization of depend(tblName.*) is blocked to keep away from producing ambiguous outcomes. As a result of depend(*) and depend(tblName.*) will output in another way if there's any null values. To revive the habits earlier than Spark 3.2, you'll be able to set spark.sql.legacy.allowStarWithSingleTableIdentifierInCount to true.

In Spark 3.0, unfavorable scale of decimal will not be allowed by default, for instance, knowledge kind of literal like 1E10BD is DecimalType(11, 0). In Spark model 2.4 and under, it was DecimalType(2, -9). To revive the habits earlier than Spark 3.0, you'll be able to set spark.sql.legacy.allowNegativeScaleOfDecimal to true.

As seen within the up to date Glue 4.0 (Spark 3.3.0) script diff in comparison with the Glue 2.0 (Spark 2.4.3) script and the ensuing improve abstract, a complete of six totally different code and configuration updates had been utilized throughout the six makes an attempt of the Spark Improve Evaluation.

• Try #1 included a Spark SQL configuration (spark.sql.adaptive.enabled) to revive the appliance habits as a brand new characteristic for Spark SQL adaptive question execution is launched beginning Spark 3.2. Customers can examine this configuration change and may additional allow or disable it as per their choice.
• Try #2 resolved a Python language change between Python 3.7 and three.10 with the introduction of a brand new summary base class (abc) below the Python collections module for importing Sequence.
• Try #3 resolved an error encountered on account of a change in habits of DataFrame API beginning Spark 3.1 the place path choice can’t exist with different DataFrameReader operations.
• Try #4 resolved an error brought on by a change within the Spark SQL perform API signature for DATE_ADD which now solely accepts integers because the second argument ranging from Spark 3.0.
• Try #5 resolved an error encountered because of the change in habits Spark SQL perform API for depend(tblName.*) beginning Spark 3.2. The habits was restored with the introduction of a brand new Spark SQL configuration spark.sql.legacy.allowStarWithSingleTableIdentifierInCount
• Try #6 efficiently accomplished the evaluation and ran the brand new script on Glue 4.0 with none new errors. The ultimate try resolved an error encountered because of the prohibited use of unfavorable scale for forged(DecimalType(3, -6) in Spark DataFrame API beginning Spark 3.0. The difficulty was addressed by enabling the brand new Spark SQL configuration spark.sql.legacy.allowNegativeScaleOfDecimal.

Essential issues for preview

As you start utilizing automated Spark upgrades through the preview interval, there are a number of essential features to think about for optimum utilization of the service:

• Service scope and limitations – The preview launch focuses on PySpark code upgrades from AWS Glue variations 2.0 to model 4.0. On the time of writing, the service handles PySpark code that doesn’t depend on further library dependencies. You possibly can run automated upgrades for as much as 10 jobs concurrently in an AWS account, permitting you to effectively modernize a number of jobs whereas sustaining system stability.
• Optimizing prices through the improve course of – As a result of the service makes use of generative AI to validate the improve plan by means of a number of iterations, with every iteration working as an AWS Glue job in your account, it’s important to optimize the validation job run configurations for cost-efficiency. To attain this, we advocate specifying a run configuration when beginning an improve evaluation as follows:
  • Utilizing non-production developer accounts and choosing pattern mock datasets that characterize your manufacturing knowledge however are smaller in measurement for validation with Spark Upgrades.
  • Utilizing right-sized compute sources, comparable to G.1X staff, and choosing an applicable variety of staff for processing your pattern knowledge.
  • Enabling Glue auto scaling when relevant to routinely alter sources based mostly on workload.

  For instance, in case your manufacturing job processes terabytes of information with 20 G.2X staff, you would possibly configure the improve job to course of a couple of gigabytes of consultant knowledge with 2 G.2X staff and auto scaling enabled for validation.

• Preview finest practices – Through the preview interval, we strongly advocate beginning your improve journey with non-production jobs. This method lets you familiarize your self with the improve workflow, and perceive how the service handles various kinds of Spark code patterns.

Your expertise and suggestions are essential in serving to us improve and enhance this characteristic. We encourage you to share your insights, ideas, and any challenges you encounter by means of AWS Help or your account workforce. This suggestions will assist us enhance the service and add capabilities that matter most to you throughout preview.

Conclusion

This publish demonstrates how automated Spark upgrades can help with migrating your Spark functions in AWS Glue. It simplifies the migration course of through the use of generative AI to routinely establish the mandatory script modifications throughout totally different Spark variations.

To be taught extra about this characteristic in AWS Glue, see Generative AI upgrades for Apache Spark in AWS Glue.

A particular because of everybody who contributed to the launch of generative AI upgrades for Apache Spark in AWS Glue: Shuai Zhang, Mukul Prasad, Liyuan Lin, Rishabh Nair, Raghavendhar Thiruvoipadi Vidyasagar, Tina Shao, Chris Kha, Neha Poonia, Xiaoxi Liu, Japson Jeyasekaran, Suthan Phillips, Raja Jaya Chandra Mannem, Yu-Ting Su, Neil Jonkers, Boyko Radulov, Sujatha Rudra, Mohammad Sabeel, Mingmei Yang, Matt Su, Daniel Greenberg, Charlie Sim, McCall Petier, Adam Rohrscheib, Andrew King, Ranu Shah, Aleksei Ivanov, Bernie Wang, Karthik Seshadri, Sriram Ramarathnam, Asterios Katsifodimos, Brody Bowman, Sunny Konoplev, Bijay Bisht, Saroj Yadav, Carlos Orozco, Nitin Bahadur, Kinshuk Pahare, Santosh Chandrachood, and William Vambenepe.

In regards to the Authors

Noritaka Sekiyama is a Principal Large Information Architect on the AWS Glue workforce. He’s accountable for constructing software program artifacts to assist prospects. In his spare time, he enjoys biking along with his new street bike.

Keerthi Chadalavada is a Senior Software program Improvement Engineer at AWS Glue, specializing in combining generative AI and knowledge integration applied sciences to design and construct complete options for patrons’ knowledge and analytics wants.

Shubham Mehta is a Senior Product Supervisor at AWS Analytics. He leads generative AI characteristic improvement throughout providers comparable to AWS Glue, Amazon EMR, and Amazon MWAA, utilizing AI/ML to simplify and improve the expertise of information practitioners constructing knowledge functions on AWS.

Pradeep Patel is a Software program Improvement Supervisor on the AWS Glue workforce. He’s obsessed with serving to prospects clear up their issues through the use of the ability of the AWS Cloud to ship extremely scalable and sturdy options. In his spare time, he likes to hike and play with internet functions.

Chuhan Liu is a Software program Engineer at AWS Glue. He’s obsessed with constructing scalable distributed programs for giant knowledge processing, analytics, and administration. He’s additionally eager on utilizing generative AI applied sciences to supply brand-new expertise to prospects. In his spare time, he likes sports activities and enjoys taking part in tennis.

Vaibhav Naik is a software program engineer at AWS Glue, obsessed with constructing sturdy, scalable options to deal with complicated buyer issues. With a eager curiosity in generative AI, he likes to discover modern methods to develop enterprise-level options that harness the ability of cutting-edge AI applied sciences.

Mohit Saxena is a Senior Software program Improvement Supervisor on the AWS Glue and Amazon EMR workforce. His workforce focuses on constructing distributed programs to allow prospects with simple-to-use interfaces and AI-driven capabilities to effectively remodel petabytes of information throughout knowledge lakes on Amazon S3, and databases and knowledge warehouses on the cloud.