Modèles Structured Streaming sur Databricks

Ceci contient des Notebooks et des exemples de code pour les modèles courants de travail avec Structured Streaming sur Databricks.

Prise en main de Structured Streaming

Si vous débutez avec Structured Streaming, consultez Exécuter votre première charge de travail Structured Streaming.

Écrire dans Cassandra en tant que récepteur pour Structured Streaming en Python

Apache Cassandra est une base de données OLTP distribuée, à faible latence, évolutive et hautement disponible.

Structured Streaming fonctionne avec Cassandra via le connecteur Spark Cassandra. Ce connecteur prend en charge les API RDD et DataFrame, et il offre un support natif pour l'écriture de données en streaming. Important Vous devez utiliser la version correspondante du spark-cassandra-connector-assembly.

L'exemple suivant se connecte à un ou plusieurs hôtes d'un cluster de base de données Cassandra. Il spécifie également les configurations de connexion telles que l'emplacement du point de contrôle et les noms spécifiques des keyspace et des tables :

Python
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()

Écrire dans Azure Synapse Analytics avec `foreachBatch()` en Python

streamingDF.writeStream.foreachBatch() vous permet de réutiliser les enregistreurs de données batch existants pour écrire la sortie d'une query streaming dans Azure Synapse Analytics. Consultez la documentation foreachBatch pour plus de détails.

Pour exécuter cet exemple, vous avez besoin du connecteur Azure Synapse Analytics. Pour plus de détails sur le connecteur Azure Synapse Analytics, consultez Interroger les données dans Azure Synapse Analytics.

Python
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()
    )

Écrire dans Amazon DynamoDB en utilisant `foreach()` en Scala et Python

streamingDF.writeStream.foreach() vous permet d'écrire la sortie d'une query de streaming vers des emplacements arbitraires.

Utiliser Python

Cet exemple montre comment utiliser streamingDataFrame.writeStream.foreach() en Python pour écrire dans DynamoDB. La première étape récupère la ressource Boto DynamoDB. Cet exemple est écrit pour utiliser access_key et secret_key, mais Databricks vous recommande d'utiliser des profils d'instance. Voir Didacticiel : Configurer l'accès à S3 avec un profil d'instance.

Définissez quelques méthodes auxiliaires pour créer une table DynamoDB pour l'exécution de l'exemple.

Python
table_name = "PythonForeachTest"

def get_dynamodb():
  import boto3

  access_key = "<access key>"
  secret_key = "<secret key>"
  region = "<region name>"
  return boto3.resource('dynamodb',
                 aws_access_key_id=access_key,
                 aws_secret_access_key=secret_key,
                 region_name=region)

def createTableIfNotExists():
    '''
    Create a DynamoDB table if it does not exist.
    This must be run on the Spark driver, and not inside foreach.
    '''
    dynamodb = get_dynamodb()

    existing_tables = dynamodb.meta.client.list_tables()['TableNames']
    if table_name not in existing_tables:
      print("Creating table %s" % table_name)
      table = dynamodb.create_table(
          TableName=table_name,
          KeySchema=[ { 'AttributeName': 'key', 'KeyType': 'HASH' } ],
          AttributeDefinitions=[ { 'AttributeName': 'key', 'AttributeType': 'S' } ],
          ProvisionedThroughput = { 'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5 }
      )

      print("Waiting for table to be ready")

table.meta.client.get_waiter('table_exists').wait(TableName=table_name)

Définissez les classes et les méthodes qui écrivent dans DynamoDB et appelez-les ensuite depuis foreach. Il existe deux façons de spécifier votre logique personnalisée dans foreach.

Utilisez une fonction : c'est l'approche simple qui peut être utilisée pour écrire une ligne à la fois. Cependant, l'initialisation client/connexion pour écrire une ligne sera effectuée à chaque appel.

Python
def sendToDynamoDB_simple(row):
  '''
  Function to send a row to DynamoDB.
  When used with `foreach`, this method is going to be called in the executor
  with the generated output rows.
  '''
  # Create client object in the executor,
  # do not use client objects created in the driver
  dynamodb = get_dynamodb()

  dynamodb.Table(table_name).put_item(
      Item = { 'key': str(row['key']), 'count': row['count'] })

Utilisez une classe avec les méthodes open, process et close : Ceci permet une implémentation plus efficace où un client/une connexion est initialisé(e) et plusieurs lignes peuvent être écrites.

Python
class SendToDynamoDB_ForeachWriter:
  '''
  Class to send a set of rows to DynamoDB.
  When used with `foreach`, copies of this class is going to be used to write
  multiple rows in the executor. See the python docs for `DataStreamWriter.foreach`
  for more details.
  '''

  def open(self, partition_id, epoch_id):
    # This is called first when preparing to send multiple rows.
    # Put all the initialization code inside open() so that a fresh
    # copy of this class is initialized in the executor where open()
    # will be called.
    self.dynamodb = get_dynamodb()
    return True

  def process(self, row):
    # This is called for each row after open() has been called.
    # This implementation sends one row at a time.
    # For further enhancements, contact the Spark+DynamoDB connector
    # team: https://github.com/audienceproject/spark-dynamodb
    self.dynamodb.Table(table_name).put_item(
        Item = { 'key': str(row['key']), 'count': row['count'] })

  def close(self, err):
    # This is called after all the rows have been processed.
    if err:
      raise err

Invoquez foreach dans votre query de streaming avec la fonction ou l’objet ci-dessus.

Python
from pyspark.sql.functions import *

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
    .foreach(SendToDynamoDB_ForeachWriter())
    #.foreach(sendToDynamoDB_simple)  // alternative, use one or the other
    .outputMode("update")
    .start()
)

Utiliser Scala

Cet exemple montre comment utiliser streamingDataFrame.writeStream.foreach() en Scala pour écrire dans DynamoDB.

Pour exécuter ceci, vous devrez créer une table DynamoDB qui possède une seule clé de chaîne de caractères nommée « value ».

Définissez une implémentation de l'interface ForeachWriter qui effectue l'écriture.

Scala
import org.apache.spark.sql.{ForeachWriter, Row}
import com.amazonaws.AmazonServiceException
import com.amazonaws.auth._
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder
import com.amazonaws.services.dynamodbv2.model.AttributeValue
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException
import java.util.ArrayList

import scala.collection.JavaConverters._

class DynamoDbWriter extends ForeachWriter[Row] {
  private val tableName = "<table name>"
  private val accessKey = "<aws access key>"
  private val secretKey = "<aws secret key>"
  private val regionName = "<region>"

  // This will lazily be initialized only when open() is called
  lazy val ddb = AmazonDynamoDBClientBuilder.standard()
    .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey)))
    .withRegion(regionName)
    .build()

  //
  // This is called first when preparing to send multiple rows.
  // Put all the initialization code inside open() so that a fresh
  // copy of this class is initialized in the executor where open()
  // will be called.
  //
  def open(partitionId: Long, epochId: Long) = {
    ddb  // force the initialization of the client
    true
  }

  //
  // This is called for each row after open() has been called.
  // This implementation sends one row at a time.
  // A more efficient implementation can be to send batches of rows at a time.
  //
  def process(row: Row) = {
    val rowAsMap = row.getValuesMap(row.schema.fieldNames)
    val dynamoItem = rowAsMap.mapValues {
      v: Any => new AttributeValue(v.toString)
    }.asJava

    ddb.putItem(tableName, dynamoItem)
  }

  //
  // This is called after all the rows have been processed.
  //
  def close(errorOrNull: Throwable) = {
    ddb.shutdown()
  }
}

Utilisez le DynamoDbWriter pour écrire un Stream de débit dans DynamoDB.

Scala
spark.readStream
  .format("rate")
  .load()
  .select("value")
  .writeStream
  .foreach(new DynamoDbWriter)
  .start()

Jointures Stream-Stream

Ces deux Notebooks montrent comment utiliser les jointures de Stream à Stream en Python et Scala.

Jointures Stream-Stream Notebook Python

Ouvrir le notebook dans un nouvel onglet Ouvrir dans Databricks

Jointures Stream-Stream du Notebook Scala

Ouvrir le notebook dans un nouvel onglet Ouvrir dans Databricks

Prise en main de Structured Streaming​

Écrire dans Cassandra en tant que récepteur pour Structured Streaming en Python​

Écrire dans Azure Synapse Analytics avec foreachBatch() en Python​

Écrire dans Amazon DynamoDB en utilisant foreach() en Scala et Python​

Utiliser Python​

Utiliser Scala​

Jointures Stream-Stream​