一、数据仓库表结构
假设我们有两个主要表： 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语句
- 数据收集与整合
从不同数据源读取数据，假设数据存储在JSON文件中。
from pyspark.sql import SparkSessionspark = 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")
- 数据清洗
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'])
- 数据集成
# 将交易数据和客户数据根据customer_id进行关联
joined_df = transactions_df.join(customers_df, on='customer_id', how='inner')
- 数据分析
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 outmodel = 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}')