Можно ли использовать Writestream напрямую через API через spark?

Question

Я строю код на Databricks для чтения дельта-таблицы в режиме реального времени (readstream), а затем мне нужно опубликовать данные этого потока в API.
Во всех документах, которые я читаю, writestream используется только для создания файлов (. csv, .avro, .parquet, et c) или отправлены в концентратор событий. Возможно ли использовать writestream для публикации в API!?

Мой код:

from pyspark.sql.functions import unix_timestamp, round, col
import json
import pandas as pd
from pyspark.sql.functions import lit
import requests

#tried with foreach_batch but it doens't work
def foreach_batch_function(df,epochId):
    r2 = requests.post('https://demo.api.com/index.php/api/v5/smsrequest/', data=str(df), verify=False)
    r2.json()
    pass

rs = spark.readStream.format("delta").option('path','/mnt/gen2/raw/mytable').load()
df = rs.select(round('id_cliente_fat').alias('id_cliente_fat'),'fone_fat','nome_fat',unix_timestamp('dt_nasc_fat','YYYY-MM-DD').cast('timestamp').cast('date').alias('birth_date'),'email_fat')

df2 = df.selectExpr('id_cliente_fat as identifier_code','fone_fat as phone_number','nome_fat as name','birth_date','email_fat as email')

data = {'authentication':{'username':'user','password':'pass'}}
r = requests.post('https://demo.api.com/index.php/api/v5/login/', data=json.dumps(data), verify=False).json()

df3 = df2.withColumn("steps", lit("[1,2,4,7]")).withColumn("place_id", lit(164)).withColumn("token", lit(r["authentication"]["token"]))

df4 = df3.select(to_json(struct(struct("token").alias("authentication"), struct("identifier_code", "phone_number", "name", "birth_date", "email","steps","place_id").alias("smsrequest").alias("smsrequest"))).alias(""))

df4.writeStream.foreachBatch(foreach_batch_function).start() 

Answer 1

Вам необходимо передать данные в драйвер с помощью метода .collect() (это не рекомендуется для большого объема данных).

Попробуйте что-то вроде этого:

def foreach_batch_function(df,epochId):

    # Create a Json with kews the name of the columns and values the values of the df
    json_data = map(lambda row: row.asDict(), df.collect())

    r2 = requests.post('https://demo.api.com/index.php/api/v5/smsrequest/', data=json_data, verify=False)
    r2.json()
    pass