# 表格可用作变量 'df'
# 要切断的异常值数量
n_outliers = 1

import json
import ast 
import boto3
from pyspark.sql.functions import col， udf
import numpy as np
from sklearn.neighbors import NearestNeighbors

# 下面的 UDF 查询 Sagemaker 端点以获取每个图像的嵌入。
# 此示例使用 Sagemaker Jumpstart 预训练的 MobileNet 模型进行图像嵌入。
# 用我们自己的端点名称替换下面突出显示的端点名称

def get_embedding(image):
    response = boto3.client('runtime.sagemaker'，region_name='us-east-1').\
    invoke_endpoint(EndpointName='jumpstart-dft-mobilenet-v2-130-224-featurevector-4'， \
    ContentType='application/x-image'，Body=image.data)
    return json.loads(response['Body'].read())["embedding"]

convertUDF = udf(lambda z: get_embedding(z))
df_pd = df.select("image_col.path"， convertUDF(col("image_col")).alias("emb")).toPandas()

# 解析输出
df_np = []
for val in df_pd["emb"].to_list():
    df_np.append(ast.literal_eval(val))  
df_np = np.array(df_np)

# 设置最近邻树 
knn_tree = NearestNeighbors(n_neighbors=10， metric='cosine')
knn_tree.fit(df_np)

# 根据与最近 10 个邻居在嵌入空间中的距离对所有图像进行评分
scores = knn_tree.kneighbors(df_np， n_neighbors=10， return_distance=True)[0][:，1:]
scores = np.mean(scores， axis = 1) 

# 按分数排序，并将 n_outliers 个最异常的图像声明为异常值
scores_argsort = np.argsort(scores)
outliers = df_pd["path"][list(scores_argsort[-n_outliers:])].to_list()

# 接下来，让我们过滤掉异常值。我们可以在自定义转换的末尾使用以下代码段
df = df.filter(~ df.path.isin(outliers))