Aorist

Aorist 是一个 MLOps 的代码生成工具。它的目的是生成易于阅读的代码，用于数据科学中的常见重复性任务，如数据复制、常见转换以及机器学习操作。

如何构建

确保您已安装 Rust，按照这里的说明进行操作，然后运行

cargo build
cp target/debug/libaorist.so example/aorist.so

尝试使用它

./run.sh
python example/minimal.py

上面的命令会生成 Python 代码或 R 代码 -- 您可以将代码管道到文件中。

Anaconda 安装（损坏）

这些说明需要 Anaconda，并已在 Ubuntu Linux 20.04 LTS 上进行了测试。

conda create -n aorist python=3.8 anaconda
conda activate aorist
pip3 install https://storage.googleapis.com/scienz-artifacts/aorist-0.0.1-cp38-cp38-linux_x86_64.whl

Aorist 宇宙概述

假设我们正在启动一个新的项目，该项目涉及分析多个大型图数据集，例如由 SNAP 项目提供的数据集。

我们将在类似于由 Trino + MinIO 解决方案指定的 Walden 的微型数据湖中进行我们的分析。

在开始分析之前，我们希望将所有这些图复制到我们的数据湖中。在非常高的层面上，这是通过定义一个“宇宙”来实现的，这是我们在项目中关心的所有事物的总体。以下是一个这样的宇宙示例

from snap import snap_dataset
from aorist import (
    dag,
    Universe,
    ComplianceConfig,
    HiveTableStorage,
    MinioLocation,
    StaticHiveTableLayout,
    ORCEncoding,
)
from common import DEFAULT_USERS, DEFAULT_GROUPS, DEFAULT_ENDPOINTS

universe = Universe(
    name="my_cluster",
    datasets=[
        snap_dataset,
    ],
    endpoints=DEFAULT_ENDPOINTS,
    users=DEFAULT_USERS,
    groups=DEFAULT_GROUPS,
    compliance=ComplianceConfig(
        description="""
        Testing workflow for data replication of SNAP data to
        local cluster. The SNAP dataset collection is provided
        as open data by Stanford University. The collection contains
        various social and technological network graphs, with
        reasonable and systematic efforts having been made to ensure
        the removal of all Personally Identifiable Information.
        """,
        data_about_human_subjects=True,
        contains_personally_identifiable_information=False,
    ),
)

宇宙定义包含多个元素

• 我们正在讨论的数据集（关于它的更多内容稍后介绍），
• 我们可用的端点（例如，MinIO 服务器可用于存储，而不是 HDFS 或 S3 等，以及该服务器在哪里可用；我们应该使用哪个端点进行 Presto/Trino 等。）
• 将访问数据集的用户和组
• 一些合规性注释。

注意：目前用户、组和合规性注释仅作为概念证明支持 -- 这些概念不是介绍中必需的，所以我们现在会跳过它们。

生成有向无环图（DAG）

要生成一个复制我们数据的流程，我们只需运行

DIALECT = "python"
out = dag(
  universe, [
    "AllAssetsComputed",
  ], DIALECT
)

这将生成一组Python任务，这些任务将对我们的数据集中的每个资产（即每个图）执行以下操作：

• 从其远程位置下载它，
• 如果需要，解压缩它
• 移除其头信息，
• 如果需要，将文件转换为CSV格式
• 将CSV数据上传到MinIO
• 创建一个支持MinIO位置的Hive表
• 将基于CSV的Hive表转换为基于ORC的Hive表
• 删除临时基于CSV的Hive表

这组任务也被称为有向无环图（DAG）。同一个DAG可以作为Jupyter笔记本生成，例如通过设置

DIALECT = "jupyter"

或者，我们可以将DIALECT设置为"airflow"以生成Airflow DAG。

旁白：实际上发生了什么？

Aorist所做的是相当复杂的——以下是对概念细节的解释，但如果你想要更具体的内容，可以跳过这部分

• 首先，你描述宇宙。这个宇宙实际上是一个高度结构化的概念层次结构，每个概念都可以“约束”。
• 约束是“需要发生的事情”。在这个例子中，你只声明需要发生的是约束AllAssetsComputed。这个约束附加到宇宙上，宇宙是一个单例对象。
• 约束附加到特定类型的对象——有些附加到整个宇宙，有些附加到表等。
• 当其依赖的约束得到满足时，认为约束得到满足。当我们填充每个约束的依赖约束时，我们遵循一系列复杂的映射规则，尽管这些规则相当直观（但如果不进行更长的讨论就难以表达，请参阅本文档的末尾）
• 程序（“食谱”）通过驱动程序附加到这个约束图。驱动程序决定首选哪种语言（例如，驱动程序可能更喜欢Bash脚本而不是Presto等）。如果驱动程序无法为特定约束提供解决方案，它会抱怨。
• 一旦食谱附加，就会从概念层次结构中提取各种细节——例如，哪些端点要击中，实际输入数据集的模式等。
• 一旦填入各种细节，我们就有了Python代码片段的图。如果这些片段是重复的（例如，100个相同函数调用但参数不同实例），我们将它们压缩到参数字典的for循环中。
• 然后，我们将压缩的片段图进一步优化，例如，通过将重复的参数从参数字典推入for循环的主体。
• 我们还计算任务的唯一、最描述性的名称，这是约束名称和概念在层次结构中的位置的组合。（例如，wine_table_has_replicated_schema）。尽可能缩短名称，同时保持唯一性（例如，我们可能将名称缩短为wine_schema，一个更简洁的任务名称）。
• 然后，驱动程序为原生Python、Airflow或Jupyter代码生成添加脚手架。将来还将支持其他输出格式（例如，Prefect、Dagster、Makefiles等）。
• 最后，驱动程序将生成的Python AST转换为具体的字符串，然后通过Python black将其格式化为漂亮的（PEP8兼容）Python程序。

描述数据集

在我们将注意力转向我们想要用我们的数据实现的目标之前，我们需要确定数据是什么，从开始。我们通过使用Python代码创建数据集清单来实现这一点。

以下是如何为规范机器学习数据集（根据example/wine.py中的Wine数据集）创建此类清单的示例。

首先，我们定义我们的属性列表

attributes = attr_list([
    attr.Categorical("wine_class_identifier"),
    attr.PositiveFloat("alcohol"),
    attr.PositiveFloat("malic_acid"),
    attr.PositiveFloat("ash"),
    attr.PositiveFloat("alcalinity_of_ash"),
    attr.PositiveFloat("magnesium"),
    attr.PositiveFloat("total_phenols"),
    attr.PositiveFloat("non_flavanoid_phenols"),
    attr.PositiveFloat("proanthocyanins"),
    attr.PositiveFloat("color_intensity"),
    attr.PositiveFloat("hue"),
    attr.PositiveFloat("od_280__od_315_diluted_wines"),
    attr.PositiveFloat("proline"),
])

然后，我们表达一个行与一个结构体相对应，该结构体的字段定义在 attributes 列表中

wine_datum = RowStruct(
    name="wine_datum",
    attributes=attributes,
)

然后，我们声明我们的数据可以在Web上的某个地方找到，在 remote 存储中。注意，我们还记录了数据以CSV编码，以及对应单个文件的地址。这里可以注明压缩算法、头部信息等。

remote = RemoteStorage(
    location=WebLocation(
        address=("https://archive.ics.uci.edu/ml/"
                 "machine-learning-databases/wine/wine.data"),
    ),
    layout=SingleFileLayout(),
    encoding=CSVEncoding(),
)

我们需要将此数据保存在本地，在一个以ORC格式存储的Hive表中，由前缀为 wine 的MinIO位置支持。

local = HiveTableStorage(
    location=MinioLocation(name="wine"),
    layout=StaticHiveTableLayout(),
    encoding=ORCEncoding(),
)

注意以下几点

• 我们没有指定表名，因为它是自动从资产名称生成的（我们将稍后定义这一点）
• 我们声明“这个事物需要存储在MinIO中”，但在此阶段我们不关心端点。Aorist将为我们找到正确的端点并填写密钥等信息。如果MinIO不可用，它将失败。
• 这里也可以指示我们的表是静态的（即没有时间维度）还是动态的。

我们现在准备好定义我们的资产，名为 wine_table

wine_table = StaticDataTable(
    name="wine_table",
    schema=default_tabular_schema(wine_datum),
    setup=RemoteImportStorageSetup(
        tmp_dir="/tmp/wine",
        remote=remote,
        local=[local],
    ),
    tag="wine",
)

这里是我们要做什么

• 我们定义了一个名为 wine_table 的资产。这还将是创建以支持此资产（或文件、目录等）的任何Hive表的名字（根据数据集存储方式而定）。
• 我们还定义了一个模式。模式告诉我们如何将行转换为模板。例如，我们需要行中列的确切顺序，以明确知道如何将其转换为结构体。
• default_tabular_schema 是一个辅助函数，允许我们推导出这样一个模式：表中的列与结构体中的字段顺序完全相同。
• 在 setup 字段中，我们通过 RemoteImportStorageSetup 引入了“复制”远程存储的概念。这里表达的想法是我们应该确保在 remote 位置上的数据与在 local 位置上的数据完全相同（通过复制，或者如果已经可用，通过检查远程和目标数据的校验和相同等）。
• 我们还使用一个 tag 字段来帮助生成可读的任务名称和ID（例如，在Airflow中）

最后，让我们定义我们的数据集

wine_dataset = DataSet(
    name="wine",
    description="A DataSet about wine",
    sourcePath=__file__,
    datumTemplates=[wine_datum],
    assets={"wine_table": wine_table},
)

然后，可以将此数据集导入之前讨论过的宇宙中。

aside：资产/模板分离

Aorist数据集旨在包含两件事

• 数据 资产 -- 具体的信息，存储在一个或多个位置，远程、本地或某种混合安排。
• 数据 模板 -- 关于我们的数据实例（即 数据）表示的信息。

例如，一个表是一个数据资产。它有行和列，这些行和列被填充了一些可以从某个位置读取的值。

这些行和列 意味着 什么取决于模板。通常，表中的行对应于结构体，例如在典型的 dim_customers 表中。但如果我们谈论图数据，那么我们的表中的一行表示一个元组（更具体地说是一对），而不是结构体。

其他数据资产示例包括

• 包含图像文件的目录，
• 具体的机器学习模型，
• 聚合，
• 散点图，

其他数据模板的示例可能包括

• 与RGB图像对应的张量数据模板，
• 一个ML模型模板，它接受一组特定的特征（例如房屋的房间数量和面积，并产生预测，例如估值），
• 直方图数据模板，表示用于聚合的边缘列的含义，以及聚合函数（直方图中的计数）。
• 散点图模板，编码x轴和y轴的含义等。

这种概念上的区分使我们能够使用相同的模板来引用多个资产。例如，我们可能有多个具有完全相同模式的表，其中一些是包含真实数据的巨大表，而另一些则是用于开发的降采样表。这些表应使用相同的模板进行引用。

这在跟踪数据血缘方面也非常有用，在两个层面上：语义上（模板Y如何从模板X派生出来？）和具体上（表T1中的行A如何从表T2中的行B派生出来？）。

回到SNAP数据集

我们最初讨论的SNAP数据集与简单的葡萄酒数据集略有不同。一方面，它包含许多资产——这是一个用于机器学习应用的图形集合——每个图形都是其自身的资产。但行的含义保持不变：它是由标识符组成的二元组。我们通过定义模板来记录这一点

edge_tuple = IdentifierTuple(
    name="edge",
    attributes=attr_list([
        attr.NumericIdentifier("from_id"),
        attr.NumericIdentifier("to_id"),
    ]),
)

然后我们为12个数据集中的每个数据集定义一个资产。请注意，名称来自对应于每个数据集的URL模式。然而，在创建资产名称时，我们需要将破折号替换为下划线（Hive表不喜欢名称中的破折号）

names = [
    "ca-AstroPh", "ca-CondMat", "ca-GrQc", "ca-HepPh",
    "ca-HepTh", "web-BerkStan", "web-Google", "web-NotreDame",
    "web-Stanford", "amazon0302", "amazon0312", "amazon0505",
]
tables = {}
for name in names:

    name_underscore = name.replace("-", "_").lower()
    remote = RemoteStorage(
        location=WebLocation(
            address="https://snap.stanford.edu/data/%s.txt.gz" % name,
        ),
        layout=SingleFileLayout(),
        encoding=TSVEncoding(
            compression=GzipCompression(),
            header=UpperSnakeCaseCSVHeader(num_lines=4),
        ),
    )
    local = HiveTableStorage(
        location=MinioLocation(name=name_underscore),
        layout=StaticHiveTableLayout(),
        encoding=ORCEncoding(),
    )
    table = StaticDataTable(
        name=name_underscore,
        schema=default_tabular_schema(edge_tuple),
        setup=RemoteImportStorageSetup(
            tmp_dir="/tmp/%s" % name_underscore,
            remote=remote,
            local=[local],
        ),
        tag=name_underscore,
    )
    tables[name] = table

snap_dataset = DataSet(
    name="snap",
    description="The Snap DataSet",
    sourcePath=__file__,
    datumTemplates=[edge_tuple],
    assets=tables,
    tag="snap",
)

如果我们想进行机器学习呢？

作为一个概念验证，ML模型与基于表格的数据资产在实质上没有太大区别。这里有一个例子，说明我们如何声明在葡萄酒表上训练的SVM回归模型的存在

# We will train a classifier and store it in a local file.
classifier_storage = LocalFileStorage(
    location=MinioLocation(name="wine"),
    layout=SingleFileLayout(),
    encoding=ONNXEncoding(),
)
# We will use these as the features in our classifier.
features = attributes[2:10]
# This is the "recipe" for our classifier.
classifier_template = TrainedFloatMeasure(
    name="predicted_alcohol",
    comment="""
    Predicted alcohol content, based on the following inputs:
    %s
    """ % [x.name for x in features],
    features=features,
    objective=attributes[1],
    source_asset_name="wine_table",
)
# We now augment the dataset with this recipe.
wine_dataset.add_template(classifier_template)
# The classifier is computed from local data
# (note the source_asset_names dictionary)
classifier_setup = ComputedFromLocalData(
    source_asset_names={"training_dataset": "wine_table"},
    target=classifier_storage,
    tmp_dir="/tmp/wine_classifier",
)
# We finally define our regression_model as a concrete
# data asset, following a recipe defined by the template,
# while also connected to concrete storage, as defined
# by classifier_setup
regression_model = SupervisedModel(
    name="wine_alcohol_predictor",
    tag="predictor",
    setup=classifier_setup,
    schema=classifier_template.get_model_storage_tabular_schema(),
    algorithm=SVMRegressionAlgorithm(),
)
wine_dataset.add_asset(regression_model)

请注意使用命令式指令，如wine_dataset.add_asset。这是我们主要声明性语法上的一个小妥协，但它很好地映射到以下ML模型常见的思维模式

• 我们有“主要来源”，即项目外部的数据集，
• 然后通过在主要来源上迭代构建来推导出其他数据资产。

因此，常见的开发周期是在原始数据源导入后，我们向数据集中添加新的模板和资产，通过在Jupyter中首先运行Python代码，然后在本机Python中运行，然后作为Airflow任务等来微调Python代码。

请注意，虽然目前Aorist只支持生成单个文件作为DAGs，但预计未来它将支持复杂项目的多个文件生成。

SQL和派生资产

特别是当数据集是表格形式时，将数据转换视为标准SQL操作（选择、投影、分组、展开和连接）是有意义的。这些转换可以通过在创建宇宙过程中使用的derive_asset指令来支持。例如，如果我们对训练仅针对高ABV葡萄酒的模型感兴趣，我们可以编写

universe.derive_asset(
    """
    SELECT *
    FROM wine.wine_table
    WHERE wine.wine_table.alcohol > 14.0
    """,
    name="high_abv_wines",
    storage=HiveTableStorage(
        location=MinioLocation(name="high_abv_wines"),
        layout=StaticHiveTableLayout(),
        encoding=ORCEncoding(),
    ),
    tmp_dir="/tmp/high_abv_wines",
)

幕后，此指令执行两件事

• 如果需要，创建一个新模板，表示在酒精属性上过滤表的操作。
• 它创建一个新的StaticDataTable资产，位于指定的存储中。此表将在其源表（即FROM子句中的表）准备好后才进行计算。

如何构建

构建货物库（需要安装Rust）

cargo build

尝试Python代码针对.so库

./run.sh

重建pip wheel（需要maturin）

maturin build

内部

Aorist的“秘密配方”是一个Rust核心。尽管我们通常通过Python与Aorist交互，但深层依赖Rust的效率和类型安全性来帮助我们处理数据任务令人印象深刻的复杂性。以下注释涉及Aorist的一些核心概念。它们仍然是正在进行的工作。

概念

Aorist使用两种类型的概念

• 抽象概念（例如“位置”）
• 具体的概念（例如，“一个Google Cloud Storage位置”，或者GCSLocation）。

抽象概念之间的关系代表了Aorist提供的核心语义模型。这不应该频繁变化。理想情况下，这不应该发生变化。

具体概念“实例化”抽象概念，就像在面向对象语言中类可以实例化特性或接口一样（实际上，这是Aorist中具体概念实现的原理）。

抽象概念具有以下层次结构

• Universe：Aorist的核心抽象，每个项目一个
  • DataSet：一组具有相互关联模式的实例化对象
  • User：访问对象的个人
  • Group：一组用户
  • Roles：用户访问数据的方式
  • RoleBindings：用户和角色之间的连接

以下是Aorist概念的当前层次结构

约束

约束是关于概念可以验证的事实。约束可能具有依赖约束。例如，我们可能对Universe有“是一致的”约束，这进一步分解为

• “数据集是复制的”，
• “用户被实例化”，
• “角色绑定被创建”，
• “数据访问策略得到执行”。

依赖约束只是告诉我们，为了使约束为真，需要满足什么条件。依赖约束可以定义在同一概念上，在依赖概念上，或在更高阶概念上，但不能形成一个循环。因此，我们不能说约束A依赖于B，B依赖于C，C又依赖于A。

这听起来相当枯燥。下面是一个示例约束集的图，以帮助更好地可视化正在发生的事情

当依赖约束定义在较低阶概念上时，我们将认为当所有与直接从约束概念继承的较低阶概念相关的依赖约束都已满足时，依赖性得到满足。

例如，我们可能说，对宇宙（我们用于数据仓库或数据湖的抽象）的约束：“没有列包含PII”要在所有表的所有列都确认不包含任何PII时得到满足。

当依赖约束定义在较高阶概念上时，我们将认为当放置在确切较高阶祖先上的依赖约束得到满足时，依赖性得到满足。

因此，例如，如果我们可以确认仓库中没有任何数据包含PII，那么基于数据仓库的数据训练的模型可以发布到网上。这是一个非常严格的保证，但它是逻辑上正确的——如果仓库中没有PII，模型中就不会有PII。这就是为什么我们可以在模型级别有一个依赖于宇宙级别的“没有PII”约束的原因。

约束DAG生成

约束和概念都在一个非常抽象的层面上操作。它们是我们理解我们关注的数据流中事物的基本语义构建块。但我们的YAML文件将定义概念的“实例”，即Aorist的对象。StaticDataTable是一个概念，但我们可能有200个静态数据表，我们希望对这些表施加相同的约束。例如，我们希望所有这些表都经过审计等。

回顾上述概念层次结构，我们通过“遍历”DAG的概念（黑色）和约束（红色）部分，将约束DAG转换为ETL管道的原型。

以下是约束DAG的形状

[1]（注意：将来我们将支持约束上的过滤器，但现在假设所有约束都必须对所有实例成立）。

关于此DAG的一些注意事项

• 它包含一些多余的依赖项，例如 DownloadDataFromRemote 和 ReplicatedData 之间的依赖关系。
• 一些约束纯粹是“装饰性的”-- DataFromRemoteDownloaded 实际上只是 DownloadDataFromRemote 的包装器，将其“提升”到根级别，以便 UploadDataToLocal 可以依赖它。

程序

约束可以在两个阶段满足

1. 首先，必须满足任何依赖约束。
2. 然后，与约束关联的程序必须成功运行。

程序是实际数据操作发生的地方。程序示例包括：“将此数据从A移动到B”，“训练此模型”或“匿名化此数据”等。程序以模板的形式编写，具有对实例化对象层次结构的完全访问权限。

程序是用一种“方言”编写的，该方言包含被认为是有效代码的内容。例如，“Python3与numpy和PyTorch”就是一种方言。对于Python方言，我们可能会将conda requirements.txt 文件或Docker镜像附加到方言，等等。对于R方言，我们可能会附加一个库列表和R版本，或一个Docker镜像。

驱动程序

注意，可能存在多个程序可以技术上满足一个约束。一个 驱动程序 决定应用哪个程序（给定偏好顺序），并负责将其实例化为在特定部署中运行的有效代码。例如，驱动程序可能负责将约束图转换为在特定数据部署中运行的有效的Airflow代码等。

测试

cargo test --no-default-features

注意：请确保libpython已添加到您的 LD_LIBRARY_PATH。例如。

export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/home/bogdan/anaconda3/lib/