如何使延迟加载 Apache Spark Dataframe 连接到 REST API

Posted 2023-04-18

【中文标题】如何使延迟加载 Apache Spark Dataframe 连接到 REST API

【英文标题】：How to make a lazy loading Apache Spark Dataframe connected to a REST API

【发布时间】：2018-11-29 04:39:18

【问题描述】：

我很高兴 Spark 允许我与数据库表建立 JDBC 连接，然后在其上构建转换，直到触发评估。我想用 REST API 连接做同样的事情。从理论上讲，这将提供一种在逻辑视图中集成数据库和 API 信息的方法。是否可以将 Spark 数据帧绑定到自定义函数，该函数使用延迟评估生成的参数调用 API？

这里有一些可玩的 pySpark 代码：

import findspark, json, requests
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession \
    .builder \
    .appName("basic test") \
    .getOrCreate()

url = 'http://worldclockapi.com/api/json/utc/now'
headers = "accept": "application/json;charset=UTF-8"
results = requests.get(url, headers=headers)
obj = json.loads(results.text)
mydict = k:v for k,v in obj.items() if v is not None
df = spark.createDataFrame([mydict])
df.show()

在这个例子中，API 调用不是惰性的，也不是由 show() 触发的。

我知道这很容易通过 pySpark API 实现。可以在 Scala 中完成吗？是否有可以在 Spark 中执行此操作的软件包？

可能与In Apache Spark, how to make an RDD/DataFrame operation lazy?相关

【参考方案1】：

可以延迟调用 REST API，但您需要将其放在 map 函数中（在处理 RDD 时）或 UDF（在 Dataframe API 中）：

>>> from pyspark.sql import Row
>>> from pyspark.sql.functions import *
>>> import requests
>>> 
>>> urls = [Row(url='http://worldclockapi.com/api/json/utc/now')] * 10
>>> call_time_api = lambda url: requests.get(url).json()['currentFileTime']
>>> 
>>> spark.createDataFrame(urls) \
...     .withColumn('time', udf(call_time_api)('url')) \
...     .show(truncate=False)
+-----------------------------------------+------------------+                  
|url                                      |time              |
+-----------------------------------------+------------------+
|http://worldclockapi.com/api/json/utc/now|131879608910925580|
|http://worldclockapi.com/api/json/utc/now|131879608911081830|
|http://worldclockapi.com/api/json/utc/now|131879608911238454|
|http://worldclockapi.com/api/json/utc/now|131879608911550881|
|http://worldclockapi.com/api/json/utc/now|131879608911706855|
|http://worldclockapi.com/api/json/utc/now|131879608911706855|
|http://worldclockapi.com/api/json/utc/now|131879608911863229|
|http://worldclockapi.com/api/json/utc/now|131879608912019732|
|http://worldclockapi.com/api/json/utc/now|131879608912175607|
|http://worldclockapi.com/api/json/utc/now|131879608912175607|
+-----------------------------------------+------------------+

事实上，这对于通过分页废弃 API 结果非常有用 - 首先您创建一个 URL 数组（每个 URL 用于不同的结果页面），然后您可以在 spark 上下文中并行检索数据并创建结果的 Dataframe。