pyspark - Not able to create a redshift table using glue - Stack Overflow

admin管理员组

文章数量:1332377

Getting error: exception: java.sql.SQLException: Exception thrown in awaitResult:

the table is getting created flight_details but there is only one column dummy in it and the schema defined in create table sql query is not there. So please help or if there is any way to create a empty table using glue in amazon redshift.

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

## @params: [JOB_NAME]
job_name = os.environ.get('AWS_GLUE_JOB_NAME', 'testing1')
args = getResolvedOptions(sys.argv, ['JOB_NAME','TempDir'])
temp_dir = args['TempDir']
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

jobmit()
#temp_dir = "s3://prateekproject1/temp_dir/"
redshift_url = "jdbc:redshift://flighttest.ckbw5q2qccz6.eu-north-1.redshift.amazonaws:5439/dev"
redshift_user = "awsuser"
redshift_password = "Prateek1997"
redshift_table = "flight.flight_details"
create_table_sql = f"""
    CREATE TABLE IF NOT EXISTS {redshift_table}(
    Carrier varchar(100),
    OriginAirportID INT,
    DestAirportID   INT,
    DepDelay    INT,
    ArrDelay    INT
    )
    DISTSTYLE KEY
    DISTKEY(OriginAirportID);
 """
 
try: 
    empty_df = spark.createDataFrame([], schema="dummy STRING")
    spark._jvm.java.lang.Class.forName("com.amazon.redshift.jdbc42.Driver")
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(empty_df, glueContext, "empty_frame"),
        connection_type="redshift",
        connection_options={
            "url": redshift_url,
            "user": redshift_user,
            "password": redshift_password,
            "preactions": create_table_sql,  # Executes the CREATE TABLE query before loading data
            "dbtable": "flight.flight_details",  # The table name
            "redshiftTmpDir": temp_dir  # Temporary directory for Redshift interactions
            }
        )
    database_name = "flight"  # Replace with the database name
    table_name = "prateekproject1"
    df = glueContext.create_dynamic_frame.from_catalog(
    database = database_name,
    table_name = table_name
    )
    df.printSchema()
    dataframe = df.toDF()
    print("Sample Data:")
    dataframe.show(10)
    print("Table created succesfully")
except Exception as e:
    print(e)
jobmit()````

Getting error: exception: java.sql.SQLException: Exception thrown in awaitResult:

the table is getting created flight_details but there is only one column dummy in it and the schema defined in create table sql query is not there. So please help or if there is any way to create a empty table using glue in amazon redshift.

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

## @params: [JOB_NAME]
job_name = os.environ.get('AWS_GLUE_JOB_NAME', 'testing1')
args = getResolvedOptions(sys.argv, ['JOB_NAME','TempDir'])
temp_dir = args['TempDir']
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

jobmit()
#temp_dir = "s3://prateekproject1/temp_dir/"
redshift_url = "jdbc:redshift://flighttest.ckbw5q2qccz6.eu-north-1.redshift.amazonaws:5439/dev"
redshift_user = "awsuser"
redshift_password = "Prateek1997"
redshift_table = "flight.flight_details"
create_table_sql = f"""
    CREATE TABLE IF NOT EXISTS {redshift_table}(
    Carrier varchar(100),
    OriginAirportID INT,
    DestAirportID   INT,
    DepDelay    INT,
    ArrDelay    INT
    )
    DISTSTYLE KEY
    DISTKEY(OriginAirportID);
 """
 
try: 
    empty_df = spark.createDataFrame([], schema="dummy STRING")
    spark._jvm.java.lang.Class.forName("com.amazon.redshift.jdbc42.Driver")
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(empty_df, glueContext, "empty_frame"),
        connection_type="redshift",
        connection_options={
            "url": redshift_url,
            "user": redshift_user,
            "password": redshift_password,
            "preactions": create_table_sql,  # Executes the CREATE TABLE query before loading data
            "dbtable": "flight.flight_details",  # The table name
            "redshiftTmpDir": temp_dir  # Temporary directory for Redshift interactions
            }
        )
    database_name = "flight"  # Replace with the database name
    table_name = "prateekproject1"
    df = glueContext.create_dynamic_frame.from_catalog(
    database = database_name,
    table_name = table_name
    )
    df.printSchema()
    dataframe = df.toDF()
    print("Sample Data:")
    dataframe.show(10)
    print("Table created succesfully")
except Exception as e:
    print(e)
jobmit()````

• pyspark
• amazon-redshift
• aws-glue

Share Follow asked Nov 22, 2024 at 5:10 Prateek GoelPrateek Goel 1911 bronze badge

Add a comment  | 

2 Answers 2

Sorted by: Reset to default 0

    // delete the table if exists to ensure the changes are applied -> remove later
create_table_sql = f"""
    DROP TABLE IF EXISTS {redshift_table};
    CREATE TABLE IF NOT EXISTS {redshift_table}(
    Carrier varchar(100),
    OriginAirportID INT,
    DestAirportID   INT,
    DepDelay    INT,
    ArrDelay    INT
    )
    DISTSTYLE KEY
    DISTKEY(OriginAirportID);
 """
 
    
    hardcoded_data = [Row(GarbageColumn="hardcoded_value")]
    empty_df = spark.createDataFrame(hardcoded_data, schema=StructType([StructField("GarbageColumn", StringType(), True)]))
    
    spark._jvm.java.lang.Class.forName("com.amazon.redshift.jdbc42.Driver")
    
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(empty_df, glueContext, "empty_frame"),
        connection_type="redshift",
        connection_options={
            "url": redshift_url,
            "user": redshift_user,
            "password": redshift_password,
            "preactions": create_table_sql,  # Executes the CREATE TABLE query before loading data
            "postactions": "begin; drop table if exists flight.temp_dummy_123123123; end;" 
            "dbtable": "flight.temp_dummy_123123123",  # use a dummy name -> this will be automatically created by Glue. so make it a dummy and then delete it in 
            "redshiftTmpDir": temp_dir  # Temporary directory for Redshift interactions
            }
        )
    

pss: You can press the like button to my replies if those helped you...

The way you are using preactions look OK, however I would suggest to execute the CREATE TABLE query directly on Redshift (outside of Glue) for testing purposes and verify that it works as expected.

Another thing could be interfering is : The preactions should only execute the SQL query for creating the table. The DataFrame (empty_df) is essentially used to load empty data into Redshift, but it should not affect the schema creation.

Here is a slightly modified version of your code : Let me know if it works or we can modify it accordingly

import os
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

## @params: [JOB_NAME]
job_name = os.environ.get('AWS_GLUE_JOB_NAME', 'testing1')
args = getResolvedOptions(sys.argv, ['JOB_NAME','TempDir'])
temp_dir = args['TempDir']
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Define Redshift connection details
redshift_url = "jdbc:redshift://flighttest.ckbw5q2qccz6.eu-north-1.redshift.amazonaws:5439/dev"
redshift_user = "awsuser"
redshift_password = "Prateek1997"
redshift_table = "flight.flight_details"

# Create the table SQL statement
create_table_sql = f"""
    CREATE TABLE IF NOT EXISTS {redshift_table} (
        Carrier varchar(100),
        OriginAirportID INT,
        DestAirportID INT,
        DepDelay INT,
        ArrDelay INT
    )
    DISTSTYLE KEY
    DISTKEY(OriginAirportID);
"""

try: 
    # Create an empty DataFrame (this won't impact the schema in Redshift)
    empty_df = spark.createDataFrame([], schema="dummy STRING")

    # Load the empty DataFrame into Redshift
    spark._jvm.java.lang.Class.forName("com.amazon.redshift.jdbc42.Driver")
    
    glueContext.write_dynamic_frame.from_options(
        frame=DynamicFrame.fromDF(empty_df, glueContext, "empty_frame"),
        connection_type="redshift",
        connection_options={
            "url": redshift_url,
            "user": redshift_user,
            "password": redshift_password,
            "preactions": create_table_sql,  # Executes the CREATE TABLE query before loading data
            "dbtable": redshift_table,       # The table name
            "redshiftTmpDir": temp_dir       # Temporary directory for Redshift interactions
        }
    )

    # To verify the table schema, let's fetch and print it after the table creation
    database_name = "flight"  # Replace with the database name
    table_name = "flight_details"  # Replace with the correct table name

    df = glueContext.create_dynamic_frame.from_catalog(
        database=database_name,
        table_name=table_name
    )
    df.printSchema()

    # Convert DynamicFrame to DataFrame for further processing
    dataframe = df.toDF()
    print("Sample Data:")
    dataframe.show(10)
    print("Table created successfully")

except Exception as e:
    print(f"Error: {e}")

jobmit()

本文标签： pysparkNot able to create a redshift table using glueStack Overflow