python - 使用 PySpark 从 Neo4j 中提取数据

Question

我有一个当前存储为图表的时间序列(使用时间树结构，类似于 this )在 Neo4j 服务器实例，版本 2.3.6(所以只有 REST 接口(interface)，没有 Bolt)。我要做的是使用 以分布式方式对这些时间序列进行一些分析。 PySpark .
现在，我知道将 Spark 与 Neo4j 连接起来的现有项目，尤其是列出的 here .这些问题是他们专注于创建一个使用图形的界面。在我的情况下，图表不相关，因为我的 Neo4j Cypher 查询旨在生成值数组。下游的一切都是关于将这些数组作为时间序列处理；再次，不是图表。
我的问题是:
有没有人使用 PySpark 成功地并行查询了仅 REST 的 Neo4j 实例，如果是，您是如何做到的？
py2neo在我意识到无法跨分区共享连接对象(或者如果可以，我不知道如何)之前，库似乎是一个不错的候选者。现在我正在考虑让我的 Spark 作业在 Neo4j 服务器上运行独立的 REST 查询，但我想看看社区是如何解决这个问题的。

Answer 1

我不确定我是否完全理解这个问题。但是当我读到它时:

连接到 Neo4J 或处理任何 Neo4J 特定元素都没有问题。

您无法正确管理多个分区；没有持久性或方法来确保跨分区的通信是正确的。

如果我的理解是正确的:

我建议看(pandas) Spark UDFs .

这些与 Neo4J 无关，但我发现它们是处理任何临时 ETL 工作的好方法，其中任何 特别 可以在单个分区上进行汇总或计算。

这是一个令人惊讶的易于使用的 API，除了 UDF 中模式的先验定义可能会很痛苦。

从文档中:

import pandas as pd

from pyspark.sql.functions import pandas_udf

@pandas_udf("col1 string, col2 long")
def func(s1: pd.Series, s2: pd.Series, s3: pd.DataFrame) -> pd.DataFrame:
    s3['col2'] = s1 + s2.str.len()
    return s3

# Create a Spark DataFrame that has three columns including a sturct column.
df = spark.createDataFrame(
    [[1, "a string", ("a nested string",)]],
    "long_col long, string_col string, struct_col struct<col1:string>")

df.printSchema()
# root
# |-- long_column: long (nullable = true)
# |-- string_column: string (nullable = true)
# |-- struct_column: struct (nullable = true)
# |    |-- col1: string (nullable = true)

df.select(func("long_col", "string_col", "struct_col")).printSchema()
# |-- func(long_col, string_col, struct_col): struct (nullable = true)
# |    |-- col1: string (nullable = true)
# |    |-- col2: long (nullable = true)

需要明确的是，那些 pd.Series和 pd.DataFrame不是您必须创建的真正的 Pandas 对象。相反，用 @pandas_udf 装饰你的函数，您提供 Spark 对象，它们在每个分区中被视为 pandas 对象。
我不知道除此之外的技术细节，除了说它适用于我曾经需要尝试的任何奇怪的 UDF(如果可以先验地知道架构!)。