Python script example to load data into ADW

Author: eloilopes

data-platform/open-source-data-platforms/oci-data-flow/code-examples/load-data-to-adw.py

@@ -0,0 +1,143 @@
+# Copyright (c) 2023 Oracle and/or its affiliates.
+# 
+# The Universal Permissive License (UPL), Version 1.0
+# 
+# Subject to the condition set forth below, permission is hereby granted to any
+# person obtaining a copy of this software, associated documentation and/or data
+# (collectively the "Software"), free of charge and under any and all copyright
+# rights in the Software, and any and all patent rights owned or freely
+# licensable by each licensor hereunder covering either (i) the unmodified
+# Software as contributed to or provided by such licensor, or (ii) the Larger
+# Works (as defined below), to deal in both
+# 
+# (a) the Software, and
+# (b) any piece of software and/or hardware listed in the lrgrwrks.txt file if
+# one is included with the Software (each a "Larger Work" to which the Software
+# is contributed by such licensors),
+# without restriction, including without limitation the rights to copy, create
+# derivative works of, display, perform, and distribute the Software and make,
+# use, sell, offer for sale, import, export, have made, and have sold the
+# Software and the Larger Work(s), and to sublicense the foregoing rights on
+# either these or other terms.
+# 
+# This license is subject to the following condition:
+# The above copyright notice and either this complete permission notice or at
+# a minimum a reference to the UPL must be included in all copies or
+# substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import argparse
+import os
+
+from pyspark import SparkConf
+from pyspark.sql import SparkSession, SQLContext
+from pyspark.sql.functions import to_date
+from pyspark.sql.functions import explode
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("inputpath")
+    #parser.add_argument("--output-path", required=True)
+    args = parser.parse_args()
+
+    # Set up Spark.
+    spark = get_dataflow_spark_session()
+    sql_context = SQLContext(spark)
+
+    # Load our data.
+    df = (
+    spark.read.format("csv")
+    #.schema(schema)
+    .option("inferSchema", "true")
+    .option("header","true")
+    .option("multiLine", "false")
+    .option("delimiter",";")
+    .option("dateFormat","dd.MM.yyyy")
+    .load(args.inputpath)
+    .cache())  
+    # cache the dataset to increase computing speed
+
+    df = df.withColumn("DATE_KEY",to_date(df["DATE_KEY"],"dd.MM.yyyy"))
+
+    df.write.format("oracle") \
+    .option("adbId","<autonomous database ocid>") \
+    .option("dbtable", "<table name>") \
+    .option("user", "<username>")\
+    .option("password", "<password>")\
+    .option("truncate", "true")\
+    .mode("overwrite")\
+    .save()
+
+    # Show on the console that something happened.
+    print("Successfully saved {} rows to ADW.".format(df.count()))
+
+def get_dataflow_spark_session(
+    app_name="DataFlow", file_location=None, profile_name=None, spark_config={}
+):
+    """
+    Get a Spark session in a way that supports running locally or in Data Flow.
+    """
+    if in_dataflow():
+        spark_builder = SparkSession.builder.appName(app_name)
+    else:
+        # Import OCI.
+        try:
+            import oci
+        except:
+            raise Exception(
+                "You need to install the OCI python library to test locally"
+            )
+
+        # Use defaults for anything unset.
+        if file_location is None:
+            file_location = oci.config.DEFAULT_LOCATION
+        if profile_name is None:
+            profile_name = oci.config.DEFAULT_PROFILE
+
+        # Load the config file.
+        try:
+            oci_config = oci.config.from_file(
+                file_location=file_location, profile_name=profile_name
+            )
+        except Exception as e:
+            print("You need to set up your OCI config properly to run locally")
+            raise e
+        conf = SparkConf()
+        conf.set("fs.oci.client.auth.tenantId", oci_config["tenancy"])
+        conf.set("fs.oci.client.auth.userId", oci_config["user"])
+        conf.set("fs.oci.client.auth.fingerprint", oci_config["fingerprint"])
+        conf.set("fs.oci.client.auth.pemfilepath", oci_config["key_file"])
+        conf.set(
+            "fs.oci.client.hostname",
+            "https://objectstorage.{0}.oraclecloud.com".format(oci_config["region"]),
+        )
+        spark_builder = SparkSession.builder.appName(app_name).config(conf=conf)
+
+    # Add in extra configuration.
+    for key, val in spark_config.items():
+        spark_builder.config(key, val)
+
+    # Create the Spark session.
+    session = spark_builder.getOrCreate()
+    return session
+
+
+def in_dataflow():
+    """
+    Determine if we are running in OCI Data Flow by checking the environment.
+    """
+    if os.environ.get("HOME") == "/home/dataflow":
+        return True
+    return False
+
+
+if __name__ == "__main__":
+    main()