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()````
Share
asked Nov 22, 2024 at 5:10
Prateek GoelPrateek Goel
191 bronze badge
2 Answers
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()