Synapse notebook can use Spark structured streaming for loading data from a datalake storage to another storage. Its pretty much the same as in Databricks. Just PySpark stuff.
Firstly, mout the source. This query the source data lake been added as a linked service in Synapse first using either key or service principal or subscription, etc.
from notebookutils import mssparkutils
container_name ='your container'
account_name = 'your storage account'
mssparkutils.fs.mount(
f"abfss://{container_name}@{account_name}.dfs.core.windows.net",
"/mount",
{"linkedService":"your linked service name", "fileCacheTimeout": 5, "timeout": 5}
)
Then get the mount path
path = mssparkutils.fs.getMountPath("/mount/")
This returns something like
#'/synfs/11/mountODL/'
Reformat the path as synfs protocal
mount_path = path.replace('/synfs', 'synfs:')
#'synfs:/10/mount/'
Make sure the destination has been created and the check point folder has been created / cleared as well.
# from notebookutils import mssparkutils
# mssparkutils.fs.mkdirs('/temp/check_point')
# mssparkutils.fs.mkdirs('/raw/history')
# mssparkutils.fs.rm('/raw/history/', True)
# mssparkutils.fs.rm('/temp/check_point/', True)
Once ready:
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.getOrCreate()
history_source = mount_path + "your source folder/subfolder/" #this is accessing a mounted location
history_target = '/raw/history' #the destination storage location
history_schema = 'Tag String, TimeStamp Timestamp, value Double, IsValueGood Boolean, Error String' #the data schema
your_tags = ['your tag'] #A list of tags as tiler
checkpoint_folder = '/temp/check_point' #check point folder, so streaming to restart if needed
df = (
spark.readStream
.schema(history_schema)
.parquet(history_source )
.filter(F.col('Tag').isin(your_tags )) #load only the tags required
.withColumnRenamed('Tag', 'Name')
.withColumn('TS', F.to_timestamp('TimeStamp', 'dd-MMM-yy HH:mm:ss.S')) #the TS string format 10-MAR-20 10:53:57.1
.withColumn('Date', F.to_date(F.col('TS'), 'yyyy-MM-dd')) #extra date column for partitioning
.withColumn('Value', F.col('value').cast('double')) #casting data type
.select('Name', 'TS', 'Value', 'IsValueGood', 'Error', 'Date')
)
query = (
df.writeStream
.partitionBy('Date')
.outputMode("append")
.option("checkpointLocation",checkpoint_folder )
.format("parquet")
.option("path", history_target )
.start()
)
The weird thing is the streaming doesn't seem to be blocking the cell. It just completes and you don't know if it is still running the backend as a separate thread or not.
One way to check is use the 'query' instance returned by the writestream. It can also show query's id, name, etc.
query.recentProgress
query.lastProgress
query.status
# {'message': 'Getting offsets from FileStreamSource[synfs:/10/mountODL/raw/data/osipi/af/aquila/archive/2023/01]',
# 'isDataAvailable': False,
# 'isTriggerActive': True}
This will show the progress of the query. Also spark can show which stream query is still active
spark.streams.active
spark.streams.active[0].lastProgress
Now check the data in the destination folder
mssparkutils.fs.ls('/raw/history/')
Read into spark dataframe
df = spark.read.schema(history_schema).parquet(history_target)
df.limit(10).show()