import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext, SparkConf
from awsglue.context import GlueContext
from awsglue.job import Job
import time
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

connection_mysql8_options_source = {
    "url": "jdbc:mysql://5b9jyx2o6pbr.us-east-1.psdb.cloud/matrix-demos-jonico?workload=olap",
    "dbtable": "pixel_matrix",
    "user": "l4klb0hnyw7o",
    "password": "redacted",
    "customJdbcDriverS3Path": "s3://your-s3-bucket/mysql-connector-java-8.0.28.jar",
    "customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver"}

connection_mysql8_options_target = {
"url": "jdbc:mysql://5b9jyx2o6pbr.us-east-1.psdb.cloud/matrix-demos-jonico?workload=olap",
"dbtable": "target_matrix",
"user": "l4klb0hnyw7o",
"password": "redacted",
"customJdbcDriverS3Path": "s3://your-s3-bucket/mysql-connector-java-8.0.28.jar",
"customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver"}

print ("Starting to connect to database")
# Read from JDBC databases with custom driver
df_source = glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql8_options_source, transformation_ctx = "df_source")
df_source = ApplyMapping.apply(frame = df_source, mappings = [("id", "integer", "id", "integer"), ("cell", "string", "cell", "string"), ("pixel_data", "string", "pixel_data", "string"), ("operation", "string", "operation", "string")])

print ("Applied mapping to the Glue DynamicFrame")
df_source = df_source.drop_fields(['operation'])

df_source.printSchema()

glueContext.write_from_options(frame_or_dfc=df_source, connection_type="mysql", connection_options=connection_mysql8_options_target, transformation_ctx = "df_target")