[アップデート] Amazon SageMaker Unified Studio でカスタムビジュアルトランスフォームが利用可能になったので試してみた

% aws datazone list-domains \
    --region ap-northeast-1 \
    --query "items[].{id:id,name:name,status:status}" \
    --output table
----------------------------------------------------------------
|                          ListDomains                         |
+---------------------+---------------------------+------------+
|         id          |           name            |  status    |
+---------------------+---------------------------+------------+
|  dzd-d3w2uawk8vo7a1 |  Default_05252026_Domain  |  AVAILABLE |
+---------------------+---------------------------+------------+

% aws datazone list-projects \
    --domain-identifier dzd-d3w2uawk8vo7a1 \
    --region ap-northeast-1 \
    --query "items[].{id:id,name:name}" \
    --output table
--------------------------------------------------
|                  ListProjects                  |
+-----------------+------------------------------+
|       id        |            name              |
+-----------------+------------------------------+
|  c778yzrdr7zy55 |  admin-project-123456789012  |
+-----------------+------------------------------+

% aws datazone list-environments \
    --domain-identifier dzd-d3w2uawk8vo7a1 \
    --project-identifier c778yzrdr7zy55 \
    --region ap-northeast-1 \
    --query "items[].{id:id,name:name,status:status}" \
    --output table
---------------------------------------------------------------------------------
|                               ListEnvironments                                |
+----------------+---------------------------------------------------+----------+
|       id       |                       name                        | status   |
+----------------+---------------------------------------------------+----------+
|  crcf629vztrvh5|  AmazonSagemakerEnvironmentConfig-axosp8fvbr1dnd  |  ACTIVE  |
|  4jxormz9fjcgpl|  AmazonSagemakerEnvironmentConfig-602yfhroojlyop  |  ACTIVE  |
+----------------+---------------------------------------------------+----------+

% for E in $(aws datazone list-environments \
    --domain-identifier dzd-d3w2uawk8vo7a1 \
    --project-identifier c778yzrdr7zy55 \
    --region ap-northeast-1 --query 'items[].id' --output text); do
  echo -n "${E}: "
  aws datazone get-environment \
      --domain-identifier dzd-d3w2uawk8vo7a1 \
      --identifier "$E" \
      --region ap-northeast-1 \
      --query "provisionedResources[?name=='s3BucketPath'].value" \
      --output text
done
crcf629vztrvh5: s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared
4jxormz9fjcgpl:

% aws s3 ls s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/ --recursive
2026-05-25 02:49:28          0 shared/
2026-05-25 02:49:28        424 shared/.libs.json

% mkdir transforms

{
  "name": "mask_pii_column",
  "displayName": "Mask PII Column (Custom)",
  "description": "Masks the values of the specified column. Optionally keeps the last N characters visible.",
  "functionName": "mask_pii_column",
  "parameters": [
    {
      "name": "colName",
      "displayName": "Column name",
      "type": "str",
      "description": "Name of the column that holds the value to mask"
    },
    {
      "name": "maskChar",
      "displayName": "Mask character",
      "type": "str",
      "isOptional": true,
      "description": "Character used for masking. Defaults to *"
    },
    {
      "name": "keepLastN",
      "displayName": "Keep last N characters",
      "type": "int",
      "isOptional": true,
      "description": "Number of trailing characters to keep visible. Defaults to 0"
    }
  ]
}

from pyspark.sql import functions as F

def mask_pii_column(self, colName, maskChar="*", keepLastN=0):
    """指定したカラムの値をマスキングする。keepLastN を指定すると末尾 N 文字だけ残す。

    Visual ETL は isOptional のパラメータを未入力のまま実行しても、引数を省略せず
    空文字列を明示的に渡してくる。そのため関数定義のデフォルト引数は適用されない。
    冒頭の 2 行で既定値へ読み替えている。
    """
    maskChar = maskChar or "*"
    keepLastN = int(keepLastN or 0)

    masked_head = F.regexp_replace(
        F.expr(f"left(`{colName}`, greatest(length(`{colName}`) - {keepLastN}, 0))"),
        ".",
        maskChar,
    )
    tail = F.expr(f"right(`{colName}`, {keepLastN})")
    return self.withColumn(colName, F.concat(masked_head, tail))

transforms/
├── mask_pii_column.json
└── mask_pii_column.py

% aws s3 cp transforms/ \
    s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/transforms/ \
    --recursive \
    --exclude "*" --include "*.json" --include "*.py" \
    --region ap-northeast-1
upload: transforms/mask_pii_column.json to s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/transforms/mask_pii_column.json
upload: transforms/mask_pii_column.py to s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/transforms/mask_pii_column.py

customer_id,customer_name,phone_number,email
1001,Taro Yamada,090-1234-5678,taro.yamada@example.com
1002,Hanako Suzuki,080-9876-5432,hanako.suzuki@example.com
1003,Jiro Sato,070-1111-2222,jiro.sato@example.com
1004,Yuki Tanaka,090-3333-4444,yuki.tanaka@example.com
1005,Kenji Ito,080-5555-6666,kenji.ito@example.com
1006,Aoi Watanabe,070-7777-8888,aoi.watanabe@example.com
1007,Sora Nakamura,090-2468-1357,sora.nakamura@example.com
1008,Rin Kobayashi,080-1357-2468,rin.kobayashi@example.com

% aws s3 cp sample-data/customers.csv \
    s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/sample-data/customers.csv \
    --region ap-northeast-1
upload: sample-data/customers.csv to s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/sample-data/customers.csv

import sys
from pyspark.context import SparkContext
from pyspark.sql import SparkSession
from mask_pii_column import mask_pii_column
from pyspark.sql.dataframe import DataFrame
import pyspark.sql.functions as F

sc = SparkContext.getOrCreate()
spark = SparkSession.builder.getOrCreate()

# Script generated for node S3DataSource
S3DataSource_1785076328528 = spark.read.format("csv") \
    .option("inferschema", "true") \
    .option("multiLine", "true") \
    .option("header", "true") \
    .option("recursiveFileLookup", "false") \
    .option("sep", ",") \
    .load("s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/sample-data/customers.csv")

def is_blank_df(df):
    # Indicates if the DataFrame has no schema and no rows.
    return not df.schema.fieldNames() and not df.take(1)
def enrich_df(name, function):
    def transform_df(self, *args, **kwargs):
        if is_blank_df(self):
            return self  # No data to transform, return as is
        return function(self, *args, **kwargs)
    setattr(DataFrame, name, transform_df)
# Script generated for node CustomTransformTransform
enrich_df('mask_pii_column', mask_pii_column)
params_mask_pii_column = {"colName": "phone_number", "maskChar": "", "keepLastN": 0}
CustomTransformTransform_1785076404196 = S3DataSource_1785076328528.mask_pii_column(**params_mask_pii_column)

# Script generated for node S3DataSink
CustomTransformTransform_1785076404196.write.format("csv") \
    .option("header", True) \
    .mode("append") \
    .save("s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/output/masked-customers/")

% aws s3 cp s3://amazon-sagemaker-123456789012-ap-northeast-1-c778yzrdr7zy55/shared/output/masked-customers-fixed/part-00000-xxxxxxxx-c000.csv - \
    --region ap-northeast-1
customer_id,customer_name,phone_number,email
1001,Taro Yamada,*********5678,taro.yamada@example.com
1002,Hanako Suzuki,*********5432,hanako.suzuki@example.com
1003,Jiro Sato,*********2222,jiro.sato@example.com
1004,Yuki Tanaka,*********4444,yuki.tanaka@example.com
1005,Kenji Ito,*********6666,kenji.ito@example.com
1006,Aoi Watanabe,*********8888,aoi.watanabe@example.com
1007,Sora Nakamura,*********1357,sora.nakamura@example.com
1008,Rin Kobayashi,*********2468,rin.kobayashi@example.com