将Cipher操作应用于数据集的列

Cipher允许您对数据集的整列进行加密、解密和哈希处理。这在Pipeline Builder、Contour和Python变换中都被支持。

Pipeline Builder

Pipeline Builder是一个数据集成应用，旨在简化在Foundry中执行高质量数据集成的过程。本节演示如何在Pipeline Builder中部署Cipher操作以混淆数据集的列。要在Pipeline Builder中运行Cipher操作，用户必须拥有Cipher的数据管理许可证或管理员许可证。

加密

首先，选择Cipher加密变换。然后，选择一个表达式（要加密的列）。接下来，选择一个具有加密权限的数据管理许可证，通常在Cipher应用中先前颁发后可以在项目文件夹中找到。最后，命名输出列。

解密

首先，选择Cipher解密变换。选择表达式时，指定一个已经通过Cipher变换加密的列。对于Cipher许可证RID，选择一个具有解密权限的数据管理许可证，通常在Cipher应用中先前颁发后可以在项目文件夹中找到。注意，许可证必须是用于加密相关列的同一个Cipher通道的一部分。最后，命名输出列。

哈希

选择Cipher哈希变换。对于表达式，指定要哈希的列。接下来，从哈希Cipher通道中选择一个具有加密权限的数据管理许可证。许可证将在Cipher应用中颁发，然后保存到项目文件夹中。最后，命名输出列。

如何使用Cipher加密管道（仅限管理员和数据管理许可证）

首先，打开一个具有加密权限的Cipher许可证；许可证通常在Cipher应用中先前颁发后可以在项目文件夹中找到。然后，选择右上角的“创建管道”。

选择要加密的输入数据集，然后选择将保存管道的目标文件夹，接着选择您要加密的数据集列。注意，只有字符串列可用于加密；如果需要加密其他列，请先将其转换为字符串。选择创建管道后，Cipher将自动生成一个新管道并加密您之前选择的列。

Contour

Contour提供了一个点选用户界面，以大规模地对表格进行数据分析。本节演示如何在Contour分析中使用Cipher操作来（解）混淆数据集的列。要在Contour中运行Cipher操作，用户必须拥有Cipher的数据管理许可证或管理员许可证。首先，使用Contour工具栏的搜索模式将Cipher面板添加到分析中。

加密

要使用Cipher面板加密数据，首先选择要加密的列（列的选择顺序对操作没有影响）。接下来，选择一个具有加密权限的数据管理许可证或管理员许可证，通常在Cipher应用中先前颁发后可以在项目文件夹中找到。选择加密操作并保存面板。列值将通过此变换更新，但列名将保持不变。

解密

要使用Cipher面板解密数据，首先选择要解密的列（列的选择顺序对操作没有影响）。接下来，选择一个具有解密权限的数据管理许可证或管理员许可证，通常在Cipher应用中先前颁发后可以在项目文件夹中找到。选择解密操作并保存面板。列值将通过此变换更新，但列名将保持不变。

哈希

要使用Cipher面板哈希数据，首先选择要哈希的列（列的选择顺序对操作没有影响）。接下来，选择一个具有哈希权限的数据管理许可证或管理员许可证，通常在Cipher应用中先前颁发后可以在项目文件夹中找到。选择哈希操作并保存面板。列值将通过此变换更新，但列名将保持不变。

Python变换

设置仓库

在conda_recipe/meta.yml中的requirements.run块中添加bellaso-python-lib。您还可以通过在代码仓库环境的库面板中添加它来自动完成此操作。注意，执行变换中的Cipher操作需要管理员许可证。

加密

要加密一列，您需要在@transforms块中定义一个EncrypterInput。EncrypterInput接受Cipher许可证的RID或文件系统路径。注意Cipher许可证必须具有加密权限。

示例：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output
from pyspark.sql.functions import col
from bellaso_python_lib.encryption.encrypter_input import EncrypterInput

@transform(
    encrypter=EncrypterInput("/path/to/cipher/license"),  # 加密器的输入路径
    output=Output("/path/to/output/dataset"),  # 输出数据集的路径
    input_dataset=Input("/path/to/input/dataset")  # 输入数据集的路径
)
def encrypt_column(ctx, input_dataset, output, encrypter):
    # 使用加密器对输入数据集中的指定列进行加密
    encrypted_df = input_dataset.dataframe().withColumn("your_column_name", encrypter.dataframe().encrypt(col("your_column_name"), ctx))
    # 将加密后的数据写入输出数据集
    output.write_dataframe(encrypted_df)

解密

要解密一列，您需要在 @transforms 块中定义一个 DecrypterInput。DecrypterInput 接受 Cipher License 的 RID 或文件系统路径。请注意，Cipher License 必须具有解密权限。

示例：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output
from pyspark.sql.functions import col
from bellaso_python_lib.decryption.decrypter_input import DecrypterInput

@transform(
    decrypter=DecrypterInput("/path/to/cipher/license"),  # 加密许可证文件的路径
    output=Output("/path/to/output/dataset"),  # 输出数据集路径
    input_dataset=Input("/path/to/input/dataset")  # 输入数据集路径
)
def decrypt_column(ctx, input_dataset, output, decrypter):
    # 解密指定列并生成新的DataFrame
    decrypted_df = input_dataset.dataframe().withColumn("your_column_name", decrypter.dataframe().decrypt(col("your_column_name"), ctx))
    # 将解密后的DataFrame写入输出路径
    output.write_dataframe(decrypted_df)

以上代码实现了一个数据转换函数decrypt_column，用于从输入数据集中解密指定的列，并将结果输出到指定的路径。函数使用DecrypterInput从给定的路径获取加密许可证，读取输入数据集，解密指定列后，将解密后的DataFrame写入输出路径。

哈希

要对列进行哈希处理，您需要在 @transforms 块中定义一个 HasherInput。HasherInput 接受 Cipher License 的 RID 或文件系统路径。请注意，Cipher License 必须具有哈希权限。

示例：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from transforms.api import transform, Input, Output
from pyspark.sql.functions import col
from bellaso_python_lib.encryption.hasher_input import HasherInput

@transform(
    # 配置HasherInput，指定密钥/许可证文件路径
    hasher=HasherInput("/path/to/cipher/license"),
    # 输出数据集路径
    output=Output("/path/to/output/dataset"),
    # 输入数据集路径
    input_dataset=Input("/path/to/input/dataset")
)
def hash_column(ctx, input_dataset, output, hasher):
    # 对输入数据集中的某一列进行哈希处理
    hashed_df = input_dataset.dataframe().withColumn("your_column_name", hasher.dataframe().hash(col("your_column_name"), ctx))
    # 写入处理后的数据集
    output.write_dataframe(hashed_df)

可视化混淆

要对列进行哈希，您需要在@transforms块中定义一个EncrypterInput。EncrypterInput需要Cipher License的RID或文件系统路径。请注意，Cipher License必须具有加密权限。

示例：

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from transforms.mediasets import MediaSetInput, MediaSetOutput
import io
from bellaso_python_lib.encryption.encrypter_input import EncrypterInput
from bellaso_python_lib.types import Coordinate

@transform(
    mediaset_in=MediaSetInput("</path/to/input/media/set"),
    mediaset_out=MediaSetOutput("</path/to/output/media/set"),
    encrypter=EncrypterInput("/path/to/cipher/license"),
    polygons=Input("/path/to/polygon/dataset"),
)
def compute(mediaset_in, mediaset_out, encrypter, polygons, ctx,):

    media_references = mediaset_in.list_media_items_by_path_with_media_reference(
        ctx
    ).collect()  # noqa

    for row in media_references:
        image_file = mediaset_in.get_media_item(row["mediaItemRid"])

        plainview_image = image_file.read()

        # 在图像左上角加密一个100像素乘100像素的正方形区域。
        polygon = [Coordinate({"x": 0, "y": 0}), Coordinate({"x": 100, "y": 0}), Coordinate({"x": 100, "y": 100}), Coordinate({"x": 0, "y": 100})]

        polygon_list = [polygon]

        if plainview_image:
            encrypted_image = encrypter.encrypt_image(plainview_image, polygon_list, ctx)

            mediaset_out.put_media_item(io.BytesIO(encrypted_image), row["path"])

在这段代码中，我们定义了一个名为 compute 的函数，用于处理媒体集中的图像数据。这个函数使用了加密功能来加密图像的指定区域。在循环中，我们读取每个图像文件，并在其左上角加密一个 100 像素乘 100 像素的正方形区域。最后，将加密后的图像存储到输出媒体集中。