Thanks for getting back to me. I'm not fully convinced yet 🙂
Copying data after writing it is not an option.
With some custom code I was able to obtain an access token from OneLake using SNI. I then passed the token to Spark. I'm now stuck with OneLake reporting a 404, stating the path I want to access would not exist, while it does.
Notice that the approach below is not using linked services to obtain an access token.
* Is there a way to make this work or am I hitting a wall?
* When will Spark be supported in Synapse?
# Goal of this code: Acquire an access token using SNI for a client ID and a cert stored in a KeyVault.
# The Synapse workspace managed ID has access to the KV and can read the cert.
from cryptography.hazmat.primitives.serialization import pkcs12, Encoding, PrivateFormat, NoEncryption
from cryptography.hazmat.primitives import hashes
import base64
import msal
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
# Load PFX from Key Vault
# Tested: Client has access to workspace (is admin) and using the client and cert successfully works
# when doing a copy job in Synapse.
cert_pfx_base64 = mssparkutils.credentials.getSecretWithLS("My_KeyVault_LinkedService", "MyCertificateName")
cert_pfx_bytes = base64.b64decode(cert_pfx_base64)
# Extract components
private_key, certificate, _ = pkcs12.load_key_and_certificates(cert_pfx_bytes, b"")
private_key_pem = private_key.private_bytes(
encoding=Encoding.PEM,
format=PrivateFormat.PKCS8,
encryption_algorithm=NoEncryption()
).decode()
cert_pem = certificate.public_bytes(Encoding.PEM).decode()
thumbprint = certificate.fingerprint(hashes.SHA1()).hex()
# MSAL setup
tenant_id = "TENANT_ID_WHERE_FABRIC_WORKSPACE_LIVES"
client_id = "CLIENT_ID_USING_SNI"
app = msal.ConfidentialClientApplication(
client_id=client_id,
authority=f"https://login.microsoftonline.com/{tenant_id}",
client_credential={
"private_key": private_key_pem,
"thumbprint": thumbprint,
"public_certificate": cert_pem # <- this is what triggers SN/I
}
)
# Acquire token
result = app.acquire_token_for_client(scopes=["https://storage.azure.com/.default"])
access_token = result["access_token"]
# Checking the token: it's valid.
print(access_token[:50])
# PASS TOKEN TO SPARK
# The URL is copied from Fabric UI's properties of the folder in the Lakehouse
full_url = "abfss://4f.....@msit-onelake.dfs.fabric.microsoft.com/f....../Files"
base_url = "4f.......@msit-onelake.dfs.fabric.microsoft.com"
spark.conf.set(f"fs.azure.account.auth.type.{base_url}", "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{base_url}", "org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.access.token.{base_url}", access_token)
# CREATE TEST DATA
schema = StructType([
StructField("id", IntegerType(), False),
StructField("name", StringType(), False),
StructField("value", IntegerType(), False)
])
data = [
(1, "Alpha", 100),
(2, "Beta", 200),
(3, "Gamma", 300)
]
df = spark.createDataFrame(data, schema)
# SAVE TEST DATA
# Returns a 404 as if path would not exists, but not true. I'm also getting this if I do not pass a token at all... 😞
df.write.format("delta").mode("overwrite").save(full_url)
# Error
# An error occurred while calling o4263.save.
# : java.io.FileNotFoundException: Operation failed: "NotFound", 404, HEAD, https://msit-onelake.dfs.fabric.microsoft.com/4......../?upn=false&action=getAccessControl&timeout=90
# at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.checkException(AzureBlobFileSystem.java:1436)
# at org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem.mkdirs(AzureBlobFileSystem.java:609)
# at org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:2388)
# at org.apache.spark.sql.delta.DeltaLog.createLogDirectory(DeltaLog.scala:467)
# at org.apache.spark.sql.delta.commands.WriteIntoDelta.write(WriteIntoDelta.scala:265)
