Padrões de streaming estruturados no Azure Databricks
Isso contém blocos de anotações e exemplos de código para padrões comuns para trabalhar com Streaming Estruturado no Azure Databricks.
Introdução ao Streaming Estruturado
Se você é novo no Streaming Estruturado, consulte Executar sua primeira carga de trabalho de Streaming Estruturado.
Escreva para Cassandra como um coletor para streaming estruturado em Python
Apache Cassandra é um banco de dados OLTP distribuído, de baixa latência, escalável e altamente disponível.
O Streaming Estruturado funciona com Cassandra através do Spark Cassandra Connector. Esse conector suporta APIs RDD e DataFrame e tem suporte nativo para gravar dados de streaming. Importante Você deve usar a versão correspondente do spark-cassandra-connector-assembly.
O exemplo a seguir se conecta a um ou mais hosts em um cluster de banco de dados Cassandra. Ele também especifica configurações de conexão, como o local do ponto de verificação e o espaço de chave específico e nomes de table:
spark.conf.set("spark.cassandra.connection.host", "host1,host2")
df.writeStream \
.format("org.apache.spark.sql.cassandra") \
.outputMode("append") \
.option("checkpointLocation", "/path/to/checkpoint") \
.option("keyspace", "keyspace_name") \
.option("table", "table_name") \
.start()
Escreva no Azure Synapse Analytics usando foreachBatch() em Python
streamingDF.writeStream.foreachBatch() permite que você reutilize gravadores de dados em lote existentes para gravar a saída de uma consulta de streaming no Azure Synapse Analytics. Consulte a documentação foreachBatch para obter detalhes.
Para executar este exemplo, você precisa do conector do Azure Synapse Analytics. Para obter detalhes sobre o conector do Azure Synapse Analytics, consulte Consultar dados no Azure Synapse Analytics.
from pyspark.sql.functions import *
from pyspark.sql import *
def writeToSQLWarehouse(df, epochId):
df.write \
.format("com.databricks.spark.sqldw") \
.mode('overwrite') \
.option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
.option("forward_spark_azure_storage_credentials", "true") \
.option("dbtable", "my_table_in_dw_copy") \
.option("tempdir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>") \
.save()
spark.conf.set("spark.sql.shuffle.partitions", "1")
query = (
spark.readStream.format("rate").load()
.selectExpr("value % 10 as key")
.groupBy("key")
.count()
.toDF("key", "count")
.writeStream
.foreachBatch(writeToSQLWarehouse)
.outputMode("update")
.start()
)
Associações transmissão-transmissão
Estes dois blocos de anotações mostram como usar junções stream-stream em Python e Scala.