pyspark 如何在Spark.read?中使用Mongo DB addFields聚合管道

5cnsuln7 于 2024-01-06 发布在 Spark

关注(0)|答案(2)|浏览(172)

我尝试在Synapse Spark笔记本中使用Mongo DB aggregation pipeline。用例是将ObjectID类型字段（如_id字段）转换为$addFields字符串。
然而，我的尝试

IllegalArgumentException: Unrecognized configuration specified: (pipeline,[{ $addFields: { _id: { '$toString': '$_id' } } } ])

字符串
我一直在尝试蛮力引用标记组合，从Copilot和文档的例子看.这里是一个例子：

pipeline = "[{ '$addFields': { '_id': { '$toString': $_id } } } ]"
df = spark.read\
    .format("cosmos.olap")\
    .option("spark.synapse.linkedService", "CosmosDbMongoDb1")\
    .option("spark.cosmos.container", "<COLLECTION NAME>")\
    .option("pipeline", pipeline)\
    .load()
display(df.limit(10))

型
是我在文字格式上犯了一些低级错误，还是这是Spark连接器中缺少支持的情况？
编辑：connector docs中的这一段可以很好地为MongoDB经验丰富的人提供答案。
自定义聚合管道必须与分区器>策略兼容。例如，聚合阶段（如$group）不适用于创建多个分区的任何分区器。

pyspark

来源：https://stackoverflow.com/questions/77625996/how-to-use-mongo-db-addfields-aggregation-pipeline-in-spark-read

2条答案

按热度按时间

yrdbyhpb1#

要将ObjectID类型字段（如_id字段）转换为字符串，请遵循以下方法：
1.为将从Cosmos OLAP读取的数据定义Schema。
1.在架构中为ObjectID字段使用二进制类型。

schema = StructType([
    StructField("_rid", StringType(), True),
    StructField("_ts", LongType(), True),
    StructField("id", StringType(), True),
    StructField("_etag", StringType(), True),
    StructField("_id", StructType([
        StructField("objectId", BinaryType(), True)  # Use BinaryType for objectId
    ]), True),
    StructField("name", StructType([
        StructField("string", StringType(), True)
    ]), True),
    StructField("age", StructType([
        StructField("int32", IntegerType(), True)
    ]), True),
    StructField("marks", StructType([
        StructField("array", ArrayType(StructType([
            StructField("int32", IntegerType(), True)
        ])), True)
    ]), True),
    StructField("newAge", StructType([
        StructField("string", StringType(), True)
    ]), True),
    StructField("_partitionKey", StructType([
        StructField("string", StringType(), True)
    ]), True)
])

字符串

convert_object_id()将字节数组作为输入，并将其转换为十六进制字符串。

def convert_object_id(bytes_array):
    builder = []
    for b in bytes_array:
        builder.append(format(b, '02x'))
    return ''.join(builder)

型
1.下面的行将UDF注册到Spark，Spark DataFrames中使用。

convert_object_id_udf = udf(convert_object_id)

型
1.使用模式从CosmosOLAP读取数据，然后将数据读入DataFrame。

df = spark.read\
    .format("cosmos.olap")\
    .option("spark.synapse.linkedService", "CosmosDbMongoDb1")\
    .option("spark.cosmos.container", "secondColl")\
    .schema(schema)\
    .load()

型

dfConverted通过添加objectId和convertedObjectId创建新的DataFrame。

dfConverted = (
     df.withColumn("objectId", col("_id.objectId"))
    .withColumn("convertedObjectId", convert_object_id_udf(col("_id.objectId")))
    .select("id", "objectId", "convertedObjectId")
)

型

我尝试的代码：

from pyspark.sql.functions import udf, col
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType, IntegerType, ArrayType, BinaryType
schema = StructType([
    StructField("_rid", StringType(), True),
    StructField("_ts", LongType(), True),
    StructField("id", StringType(), True),
    StructField("_etag", StringType(), True),
    StructField("_id", StructType([
        StructField("objectId", BinaryType(), True)  # Use BinaryType for objectId
    ]), True),
    StructField("name", StructType([
        StructField("string", StringType(), True)
    ]), True),
    StructField("age", StructType([
        StructField("int32", IntegerType(), True)
    ]), True),
    StructField("marks", StructType([
        StructField("array", ArrayType(StructType([
            StructField("int32", IntegerType(), True)
        ])), True)
    ]), True),
    StructField("newAge", StructType([
        StructField("string", StringType(), True)
    ]), True),
    StructField("_partitionKey", StructType([
        StructField("string", StringType(), True)
    ]), True)
])
def convert_object_id(bytes_array):
    builder = []
    for b in bytes_array:
        builder.append(format(b, '02x'))
    return ''.join(builder)
 
convert_object_id_udf = udf(convert_object_id)
 
df = spark.read\
    .format("cosmos.olap")\
    .option("spark.synapse.linkedService", "CosmosDbMongoDb1")\
    .option("spark.cosmos.container", "secondColl")\
    .schema(schema)\
    .load()
dfConverted = (
     df.withColumn("objectId", col("_id.objectId"))
    .withColumn("convertedObjectId", convert_object_id_udf(col("_id.objectId")))
    .select("id", "objectId", "convertedObjectId")
)
 
display(dfConverted)

型

输出：

的数据

展开查看全部

赞(0）回复(0）举报 2024-01-06

ljsrvy3e2#

我在Azure文档中找到了Scala代码示例，它可以将ObjectId字段转换为字符串：

val df = spark.read.format("cosmos.olap")
.option("spark.synapse.linkedService", "xxxx")
.option("spark.cosmos.container", "xxxx")
.load()
val convertObjectId = udf((bytes: Array[Byte]) => {
    val builder = new StringBuilder
    for (b <- bytes) {
        builder.append(String.format("%02x", Byte.box(b)))
    }
    builder.toString
}
)
val dfConverted = df.withColumn("objectId", col("_id.objectId")).withColumn("convertedObjectId", convertObjectId(col("_id.objectId"))).select("id", "objectId", "convertedObjectId")
display(dfConverted)

字符串
此解决方案还回答了问题here

展开查看全部

赞(0）回复(0）举报 2024-01-06

我来回答

pyspark 如何在Spark.read?中使用Mongo DB addFields聚合管道

2条答案

相关问题

热门标签

最新问答