用PySpark和PyTorch实现跨境支付Hive数据仓库的反洗钱数据分析

一、数据仓库表结构

假设我们有两个主要表： transactions （交易表）和 customers （客户表）。

transactions 表

CREATE TABLE transactions (
    transaction_id STRING,
    customer_id STRING,
    counterparty_id STRING,
    transaction_amount DECIMAL(10, 2),
    transaction_time TIMESTAMP,
    transaction_location STRING,
    transaction_country STRING
)
PARTITIONED BY (year INT, month INT)
CLUSTERED BY (transaction_amount) INTO 4 BUCKETS;

customers 表

CREATE TABLE customers (
    customer_id STRING,
    customer_name STRING,
    customer_address STRING,
    customer_country STRING,
    risk_level INT
)

二、分析流程的PySpark语句

1. 数据收集与整合

从不同数据源读取数据，假设数据存储在JSON文件中。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Anti - Money Laundering").getOrCreate()

# 读取交易数据
transactions_df = spark.read.json("path/to/transactions.json")
# 读取客户数据
customers_df = spark.read.json("path/to/customers.json")

2. 数据清洗

from pyspark.sql.functions import col, when, row_number
from pyspark.sql.window import Window

# 去除交易数据中的重复记录
transactions_df = transactions_df.dropDuplicates()

# 处理交易数据中的缺失值，用0填充交易金额
transactions_df = transactions_df.fillna(0, subset=['transaction_amount'])

# 处理异常值，假设交易金额不能为负数，将负数标记为异常
transactions_df = transactions_df.withColumn("transaction_amount", when(col("transaction_amount") < 0, -1).otherwise(col("transaction_amount")))

# 去除客户数据中的重复记录
customers_df = customers_df.dropDuplicates()

# 处理客户数据中的缺失值，假设客户国家缺失用'Unknown'填充
customers_df = customers_df.fillna('Unknown', subset=['customer_country'])

3. 数据集成

# 将交易数据和客户数据根据customer_id进行关联
joined_df = transactions_df.join(customers_df, on='customer_id', how='inner')

4. 数据分析

from pyspark.sql.functions import count, sum, avg, col

# 设定金额阈值
amount_threshold = 100000
# 设定频率阈值，假设每天交易超过10次为频繁
frequency_threshold = 10

# 按客户ID和日期统计交易次数和总金额
daily_transaction_stats = joined_df.groupBy("customer_id", col("transaction_time").cast("date").alias("transaction_date")) \
  .agg(count("transaction_id").alias("transaction_count"), sum("transaction_amount").alias("total_amount"))

# 标记可疑交易
suspicious_transactions = daily_transaction_stats.filter((col("total_amount") > amount_threshold) | (col("transaction_count") > frequency_threshold))

三、用深度学习分析报告数据的PyTorch代码

假设我们将可疑交易数据转换为特征矩阵，用于深度学习模型训练。这里以简单的二分类（可疑或不可疑）为例。

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader

# 假设suspicious_transactions_df是上述分析得到的可疑交易数据，转换为特征矩阵
# 这里简单假设特征为交易金额和交易次数
features = torch.tensor(suspicious_transactions_df.select("total_amount", "transaction_count").collect(), dtype=torch.float32)
labels = torch.tensor(suspicious_transactions_df.select("is_suspicious").collect(), dtype=torch.float32).view(-1)

# 创建数据集和数据加载器
dataset = TensorDataset(features, labels)
dataloader = DataLoader(dataset, batch_size=32, shuffle=True)

# 定义神经网络模型
class AntiMoneyLaunderingModel(nn.Module):
    def __init__(self):
        super(AntiMoneyLaunderingModel, self).__init__()
        self.fc1 = nn.Linear(2, 16)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(16, 1)
        self.sigmoid = nn.Sigmoid()

    def forward(self, x):
        out = self.fc1(x)
        out = self.relu(out)
        out = self.fc2(out)
        out = self.sigmoid(out)
        return out

model = AntiMoneyLaunderingModel()
criterion = nn.BCELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练模型
for epoch in range(100):
    for i, (batch_features, batch_labels) in enumerate(dataloader):
        optimizer.zero_grad()
        outputs = model(batch_features)
        loss = criterion(outputs.view(-1), batch_labels)
        loss.backward()
        optimizer.step()

    if (epoch + 1) % 10 == 0:
        print(f'Epoch [{epoch + 1}/100], Loss: {loss.item():.4f}')