在pypark中作为Dynamo DB运行在AWS胶水和规则存储库中的pypark中基于规则的引擎的有效设计是什么？

如何解决在pypark中作为Dynamo DB运行在AWS胶水和规则存储库中的pypark中基于规则的引擎的有效设计是什么？

我们正在尝试为AWS在规则引擎中实现一个设计。下面是该情况的说明。

我们正在使用S3作为存储在AWS上构建一个数据湖，该存储具有不同的存储桶来表示数据充实。例如，数据进入原始存储桶。从那里选择数据并应用数据标准化（某些标准化使布尔字段统一，例如具有值0或1，将所有日期格式列转换为非格式标准，例如dd / mm / yyyy hh：mi：ss，为某些列数据等）
使用粘合作业，pyspark代码，用于编排的Step函数以及用于存储作业配置的Dynamodb和用于存储元数据的Glue目录来完成数据移动。

需要将标准化规则存储在Dynamodb中，以便胶水作业可以从DynamoDB表中选取标准化规则并将其应用于作业正在处理的数据。要处理的数据将存储在S3中，并在数据帧中提取。

动态执行此操作的最佳方法是什么？另外，该作业还应该足够灵活，以便将来在dynamoDB表中添加新规则时，应通过粘合作业将其拾取并应用于数据获取过程中（我们不想更改该作业的代码，如果添加了新规则）。规则必须在DynamoDB中，以便其他进程可以动态使用/操纵它们，并且Glue应该始终检索要在作业中使用的最新版本。

任何带有示例代码的想法都会很有帮助。谢谢！

因此，如果我总结胶水代码中的步骤，它应如下运行：

胶水作业从S3存储桶中选择一个数据表（通过目录或连接）
然后从dynamo db中选择规则。规则应包含步骤1中提取的表的规则和适用于哪一列的信息
在数据帧上使用spark sql或spark api动态应用规则
结束

解决方法

下面的链接中提供了一种指导。

In PySpark is there any way to dynamically register UDF using functions of Python class given at run time?

使用链接中提供的概念，可以开发动态规则引擎。应该处理数据的粘合作业可以读取dynamo DB中存储的配置详细信息。要存储在DynamoDB中的正确类型的配置将是规则名称，表名称，的映射（列名称和要应用的规则）。然后，在作业中，可以循环访问此配置，并且可以动态构建并在循环中运行的sql查询字符串sql。

这是一个设计问题，因此除链接中已提供的详细信息外，我没有发布其他代码。

AWS Glue具有对DynamoDB表进行读写的本机支持。对于您而言，如果只需要检索数据，则可以将DynamoDB配置为作业的源。

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv,["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"],args)

dyf = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",connection_options={
        "dynamodb.input.tableName": "test_source","dynamodb.throughput.read.percent": "1.0","dynamodb.splits": "100"
    }
)

您可以在官方文档的this page中找到有关此代码和配置的更多详细信息。

如果相反，您希望以较少的pyspark-y（但使用更多的pythonic）的方式进行管理，则可以选择导入boto3库（Python的官方AWS开发工具包）并从那里提取数据。

import boto3

dynamodb = boto3.service('dynamodb')
table = dynamodb.Table('my_table')
table.query(
  Select='ALL_ATTRIBUTES',KeyConditions={
    'my_partition_key': { "S": "some_value" }
  }
)

Here，您可以找到boto3库的官方文档。您有不同的选项来读取数据，但一般来说，应尽可能尝试在query上使用scan。

在两种情况下，您都必须确保AWs Glue具有足够的权限来对要使用的DynamoDB表执行读取和列出操作，并且在DynamoDB方面，该表具有足够的读/写输出来维持操作

在pypark中作为Dynamo DB运行在AWS胶水和规则存储库中的pypark中基于规则的引擎的有效设计是什么？

如何解决在pypark中作为Dynamo DB运行在AWS胶水和规则存储库中的pypark中基于规则的引擎的有效设计是什么？

解决方法

相关推荐