df_Bronze_Insurance_Data.write.format("json").option("mode","overwrite").save("oci://test-demo@OSNamespace/insur_claim/claim_curated")
Reference: https://github.com/oracle-samples/oracle-dataflow-samples/tree/main/python/loadadw
%%spark
def get_authenticated_client(token_path, client, file_location=None, profile_name=None):
"""
Get an an authenticated OCI client.
Example: get_authenticated_client(token_path, oci.object_storage.ObjectStorageClient)
"""
import oci
if not in_dataflow():
# We are running locally, use our API Key.
if file_location is None:
file_location = oci.config.DEFAULT_LOCATION
if profile_name is None:
profile_name = oci.config.DEFAULT_PROFILE
config = oci.config.from_file(file_location=file_location, profile_name=profile_name)
authenticated_client = client(config)
else:
# We are running in Data Flow, use our Delegation Token.
with open(token_path) as fd:
delegation_token = fd.read()
signer = oci.auth.signers.InstancePrincipalsDelegationTokenSigner(
delegation_token=delegation_token
)
authenticated_client = client(config={}, signer=signer)
return authenticated_client
def get_password_from_secrets(token_path, password_ocid):
"""
Get a password from the OCI Secrets Service.
"""
import base64
import oci
secrets_client = get_authenticated_client(token_path, oci.secrets.SecretsClient)
response = secrets_client.get_secret_bundle(password_ocid)
base64_secret_content = response.data.secret_bundle_content.content
base64_secret_bytes = base64_secret_content.encode("ascii")
base64_message_bytes = base64.b64decode(base64_secret_bytes)
secret_content = base64_message_bytes.decode("ascii")
return secret_content
def get_delegation_token_path(spark):
"""
Get the delegation token path when we're running in Data Flow.
"""
if not in_dataflow():
return None
token_key = "spark.hadoop.fs.oci.client.auth.delegationTokenPath"
token_path = spark.sparkContext.getConf().get(token_key)
if not token_path:
raise Exception(f"{token_key} is not set")
return token_path
def get_temporary_directory():
if in_dataflow():
return "/opt/spark/work-dir/"
else:
import tempfile
return tempfile.gettempdir()
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
def download_wallet(spark, wallet_path):
"""
Download an ADW/ATP wallet file and prepare it for use in a Data Flow
application.
"""
import oci
import zipfile
# Get an object store client.
token_path = get_delegation_token_path(spark)
object_store_client = get_authenticated_client(
token_path, oci.object_storage.ObjectStorageClient
)
# Download the wallet file.
from urllib.parse import urlparse
parsed = urlparse(wallet_path)
bucket_name, namespace = parsed.netloc.split("@")
file_name = parsed.path[1:]
response = object_store_client.get_object(namespace, bucket_name, file_name)
temporary_directory = get_temporary_directory()
zip_file_path = os.path.join(temporary_directory, "wallet.zip")
with open(zip_file_path, "wb") as fd:
for chunk in response.data.raw.stream(1024 * 1024, decode_content=False):
fd.write(chunk)
# Extract everything locally.
with zipfile.ZipFile(zip_file_path, "r") as zip_ref:
zip_ref.extractall(temporary_directory)
# Distribute all wallet files.
contents = "cwallet.sso ewallet.p12 keystore.jks ojdbc.properties sqlnet.ora tnsnames.ora truststore.jks".split()
spark_context = spark.sparkContext
for file in contents:
spark_context.addFile(os.path.join(temporary_directory, file))
return temporary_directory
◉ Set below Parameters related to ADW Instance & Wallet.
%%spark
PASSWORD_SECRET_OCID = "ocid1.vaultsecret.oc1.phx.xxxxxxx"
TARGET_TABLE = "ADMIN.TB_NAME"
TNSNAME = "demolakehouseadw_medium"
USER = "admin"
WALLET_PATH = "oci://bucketname@osnamespace/Wallet_DemoLakeHouseADW.zip"
# Download and distribute our wallet file.
wallet_path = download_wallet(spark, WALLET_PATH)
adw_url = "jdbc:oracle:thin:@{}?TNS_ADMIN={}".format(TNSNAME, wallet_path)
◉ Now, Get password using the secret service.
%%spark
# Get our password using the secret service.
print("Getting wallet password")
token_path = get_delegation_token_path(spark)
password = get_password_from_secrets(token_path, PASSWORD_SECRET_OCID)
print("Done getting wallet password")
# Save the results to the database.
print("Saving processed data to " + adw_url)
properties = {
"driver": "oracle.jdbc.driver.OracleDriver",
"oracle.net.tns_admin": TNSNAME,
"password": password,
"user": USER
}
◉ Read Sample Table from ADW.
%%spark
SOURCE_TABLE = "ADMIN.RETAILPOS"
df_RetailPOS_15min = spark.read.jdbc(url=adw_url, table=SOURCE_TABLE, properties=properties)
◉ Load above Dataframe into ADW.
%%spark
#Load into ADW:
TARGET_TABLE = "ADMIN.RETAILPOS_15MINUTES"
print("TARGET_TABLE : ",TARGET_TABLE)
# Write to ADW.
print("Write to ADW : ")
df_RetailPOS_15min.write.jdbc(url=adw_url, table=TARGET_TABLE, mode="Append", properties=properties)
print("Writing done to ADW : ")
Posts
