从Amazon离子文件中提取“数据”

Question

有人使用过亚马逊量子分类器数据库(QLDB)的亚马逊离子文件吗？如果是这样的话，您知道如何提取“数据”部分来制定表格吗？也许用python来抓取数据？我试图从这些存储在s3中的文件中获取“数据”信息(我无法访问QLDB，因此无法直接查询)，然后将结果上传到Glue。

我正在尝试使用GLue执行ETL工作，但是Glue不喜欢Amazon文件，所以我需要查询这些文件中的数据，或者抓取文件中的相关信息。

谢谢。PS：我所说的“数据”信息的意思是：

{
    PersonId:"4tPW8xtKSGF5b6JyTihI1U",
    LicenseNumber:"LEWISR261LL",
    LicenseType:"Learner",
    ValidFromDate:2016–12–20,
    ValidToDate:2020–11–15
}

ref：https://docs.aws.amazon.com/qldb/latest/developerguide/working.userdata.html

Answer 1

您试过使用亚马逊离子库吗？

假设问题中提到的数据存在于一个名为"myIonFile.ion“的文件中，并且如果该文件中只有离子对象，我们可以如下所示从文件中读取数据：

from amazon.ion import simpleion

file = open("myIonFile.ion", "rb")                    # opening the file
data = file.read()                                    # getting the bytes for the file
iondata = simpleion.loads(data, single_value=False)   # Loading as ion data
print(iondata['PersonId'])                            # should print "4tPW8xtKSGF5b6JyTihI1U"

在离子食谱中提供了关于使用离子库的进一步指导

此外，我不确定您的用例，但与QLDB的交互也可以通过QLDB驱动器完成，后者直接依赖于离子库。

Answer 2

诺西西维

AWS Glue能够读取亚马逊的离子输入。但是，许多其他服务和应用程序不能，所以使用Glue将离子数据转换为JSON是个好主意。请注意，离子是一组超级JSON，向JSON中添加了一些数据类型，因此将离子转换为JSON可能会导致一些下转换。

从QLDB S3导出访问QLDB文档的一个好方法是使用Glue提取文档数据，以S3形式存储它，并使用Athena查询它。这一进程如下：

1. 将分类账数据导出到S3
2. 创建一个胶水爬行器来对导出的数据进行爬行和编目。
3. 运行一个胶水ETL作业，从导出文件中提取修订数据，将其转换为JSON，并将其写入S3。
4. 创建一个胶水爬行器来抓取和编录提取的数据。
5. 使用Athena查询提取的文档修订数据。

看看下面的PySpark脚本。它只从QLDB导出文件中提取修订的元数据和数据有效负载。

QLDB导出映射每个文档的表，但与修订数据不同。您必须做一些额外的编码才能将表名包含在输出中的修订数据中。下面的代码没有这样做，所以您将在输出中的一个表中得到所有的修订版。

还请注意，您将得到在导出的数据中发生的任何修订。也就是说，您可能会得到一个给定文档ID的多个文档修订。根据您对数据的预期使用，您可能需要弄清楚如何只获取每个文档ID的最新修订。

from awsglue.transforms import *
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from pyspark.sql.functions import explode
from pyspark.sql.functions import col
from awsglue.dynamicframe import DynamicFrame

# Initializations
sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)

# Load data.  'vehicle-registration-ion' is the name of your database in the Glue catalog for the export data.  '2020' is the name of your table in the Glue catalog.
dyn0 = glueContext.create_dynamic_frame.from_catalog(database = "vehicle-registration-ion", table_name = "2020", transformation_ctx = "datasource0")

# Only give me exported records with revisions
dyn1 = dyn0.filter(lambda line: "revisions" in line)

# Now give me just the revisions element and convert to a Spark DataFrame.
df0 = dyn1.select_fields("revisions").toDF()

# Revisions is an array, so give me all of the array items as top-level "rows" instead of being a nested array field.
df1 = df0.select(explode(df0.revisions))

# Now I have a list of elements with "col" as their root node and the revision 
# fields ("data", "metadata", etc.) as sub-elements.  Explode() gave me the "col"
# root node and some rows with null "data" fields, so filter out the nulls.
df2 = df1.where(col("col.data").isNotNull())

# Now convert back to a DynamicFrame
dyn2 = DynamicFrame.fromDF(df2, glueContext, "dyn2")

# Prep and send the output to S3
applymapping1 = ApplyMapping.apply(frame = dyn2, mappings = [("col.data", "struct", "data", "struct"), ("col.metadata", "struct", "metadata", "struct")], transformation_ctx = "applymapping1")
datasink0 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": "s3://YOUR_BUCKET_NAME_HERE/YOUR_DESIRED_OUTPUT_PATH_HERE/"}, format = "json", transformation_ctx = "datasink0")

我希望这能帮到你!