ElasticRelay 实战：MySQL 到 Elasticsearch 实时同步与数据转换脱敏保姆级教程

linjifan

751人浏览 · 2026-01-31 00:15:57

linjifan · 2026-01-31 00:15:57 发布

深入解析 Transform 引擎 —— ElasticRelay 的核心数据治理组件

前言

在现代数据架构中，将 MySQL 数据实时同步到 Elasticsearch 是一个非常常见的需求。无论是为了构建全文搜索、数据分析还是实时监控，都需要一个可靠的数据同步方案。然而，在数据同步过程中，我们往往还面临着以下挑战：

数据格式转换：MySQL 和 Elasticsearch 的数据类型不完全匹配
敏感数据脱敏：用户手机号、身份证、银行卡等敏感信息需要脱敏处理
字段重命名：源表字段名与目标索引字段名不一致
计算派生字段：需要根据现有字段计算出新的业务字段
数据过滤：只同步符合条件的数据，排除测试数据或已删除记录

ElasticRelay 的 Transform 引擎 完美解决了上述所有问题，让你可以在数据同步的同时完成复杂的数据治理任务。

ElasticRelay 简介
环境准备
MySQL CDC 配置详解
Transform 引擎核心概念
字段映射：重命名与复制
字段配置：类型转换与验证
数据脱敏：敏感信息保护
计算字段：动态派生新字段
数据过滤：精确控制同步范围
完整配置示例
启动与验证
性能调优与最佳实践
常见问题排查

1. ElasticRelay 简介

ElasticRelay 是一款高性能的 CDC（Change Data Capture）数据同步工具，专门用于将关系型数据库（MySQL、PostgreSQL）和 NoSQL 数据库（MongoDB）的数据实时同步到 Elasticsearch。

Transform 引擎是什么？

Transform 引擎是 ElasticRelay 的核心数据治理组件，它在数据从源端流向目标端的过程中，提供了强大的实时数据转换能力：

功能	描述	应用场景
字段映射	重命名、复制、移动字段	统一字段命名规范
类型转换	字符串、整数、浮点、布尔、日期等类型互转	MySQL → ES 类型适配
数据脱敏	手机号、身份证、邮箱、银行卡等敏感数据匿名化	隐私保护、合规要求
表达式计算	动态计算新字段，支持条件表达式和内置函数	业务指标衍生
数据过滤	基于条件包含/排除记录	排除测试数据、已删除记录

Transform 处理流程

MySQL Binlog Event (JSON)
         ↓
┌─────────────────────────────────────────┐
│  1. Filter    - 条件过滤                 │  ← 排除不需要的记录
│  2. Mapping   - 字段映射 (rename/copy)   │  ← 字段重命名/复制
│  3. Config    - 字段配置 (类型/排除)      │  ← 类型转换、排除字段
│  4. Masking   - 数据脱敏                 │  ← 敏感信息脱敏
│  5. Computed  - 表达式计算               │  ← 计算派生字段
└─────────────────────────────────────────┘
         ↓
Transformed Event → Elasticsearch

2. 环境准备

2.1 MySQL 配置要求

ElasticRelay 通过读取 MySQL Binlog 实现实时数据捕获，因此需要确保 MySQL 开启了 Binlog 并使用正确的格式。

检查当前 Binlog 配置：

-- 检查 Binlog 是否开启
SHOW VARIABLES LIKE 'log_bin';

-- 检查 Binlog 格式（必须为 ROW）
SHOW VARIABLES LIKE 'binlog_format';

-- 检查 Binlog 行镜像（建议为 FULL）
SHOW VARIABLES LIKE 'binlog_row_image';

-- 检查 server-id
SHOW VARIABLES LIKE 'server_id';

修改 MySQL 配置文件 (my.cnf 或 my.ini)：

[mysqld]
# 开启 Binlog
log-bin=mysql-bin

# 必须使用 ROW 格式
binlog_format=ROW

# 建议使用 FULL，记录完整的行数据
binlog_row_image=FULL

# Server ID，集群中必须唯一
server-id=1

# 可选：开启 GTID 模式（推荐）
gtid_mode=ON
enforce_gtid_consistency=ON

创建专用同步账号：

-- 创建用户
CREATE USER 'elasticrelay_user'@'%' IDENTIFIED BY 'elasticrelay_pass';

-- 授权：需要 REPLICATION 权限读取 Binlog
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'elasticrelay_user'@'%';

-- 刷新权限
FLUSH PRIVILEGES;

2.2 Elasticsearch 配置要求

确保 Elasticsearch 集群正常运行，并创建必要的索引模板（可选）：

# 检查 ES 集群健康状态
curl -u elastic:password http://localhost:9200/_cluster/health?pretty

2.3 ElasticRelay 安装

# 下载最新版本
wget https://github.com/YogooSoft/elasticrelay/tree/main/releases/download/v1.4.2/elasticrelay-linux-amd64.tar.gz

# 解压
tar -xzf elasticrelay_linux_amd64.tar.gz

# 进入目录
cd elasticrelay

# 查看目录结构
ls -la
# bin/           - 可执行文件
# config/        - 配置文件目录
# docs/          - 文档
# start.sh       - 启动脚本

3. MySQL CDC 配置详解

创建 MySQL 数据源配置文件 config/mysql_config.json：

{
  "version": "3.0",
  "data_sources": [
    {
      "id": "mysql-main",
      "name": "MySQL主数据库",
      "type": "mysql",
      "host": "localhost",
      "port": 3306,
      "user": "elasticrelay_user",
      "password": "elasticrelay_pass",
      "database": "elasticrelay",
      "server_id": 100,
      "table_filters": ["users", "orders", "products", "audit_logs"],
      "options": {
        "charset": "utf8mb4",
        "timeout": "30s",
        "binlog_format": "ROW",
        "binlog_row_image": "FULL",
        "enable_gtid": true,
        "batch_size": 1000,
        "max_connections": 10,
        "enable_performance_monitoring": true
      }
    }
  ],
  "sinks": [
    {
      "id": "es-main",
      "name": "主Elasticsearch集群",
      "type": "elasticsearch",
      "addresses": ["http://172.168.0.100:19200"],
      "user": "elastic",
      "password": "your_es_password",
      "options": {
        "index_prefix": "elasticrelay_mysql",
        "batch_size": 100,
        "flush_bytes": 1048576,
        "flush_interval": "3s",
        "request_timeout": "10s"
      }
    }
  ],
  "jobs": [
    {
      "id": "mysql-to-es-cdc",
      "name": "MySQL Binlog CDC同步任务",
      "source_id": "mysql-main",
      "sink_id": "es-main",
      "enabled": true,
      "description": "MySQL binlog CDC到Elasticsearch的实时同步任务",
      "options": {
        "initial_sync": true,
        "force_initial_sync": true,
        "batch_size": 1000,
        "checkpoint_interval": "10s",
        
        "binlog_config": {
          "start_position": "latest",
          "binlog_format": "ROW",
          "binlog_row_image": "FULL",
          "enable_gtid": true,
          "server_id": 100
        },
        
        "retry_config": {
          "max_retries": 3,
          "retry_backoff": "exponential",
          "initial_delay": "5s",
          "max_delay": "300s"
        },
        
        "monitoring": {
          "metrics_interval": "30s",
          "progress_reporting": true,
          "binlog_lag_alert_threshold": "10s"
        }
      }
    }
  ],
  "global": {
    "log_level": "info",
    "metrics_port": 8080,
    "grpc_port": 50051,
    "dlq_config": {
      "enabled": true,
      "storage_path": "./dlq",
      "max_retries": 3,
      "retry_delay": "15s"
    }
  }
}

配置项详解

配置项	说明	建议值
`server_id`	MySQL 复制 Server ID，必须唯一	100-199 范围
`table_filters`	要同步的表名列表	只包含需要的表
`enable_gtid`	启用 GTID 模式，断点续传更可靠	`true`
`batch_size`	批量处理大小	500-2000
`initial_sync`	首次启动时全量同步	`true`
`start_position`	Binlog 起始位置	`latest` 或具体位点

4. Transform 引擎核心概念

Transform 配置文件独立于数据源配置，便于管理和复用。创建 config/mysql_transform.json：

配置文件基本结构

{
  "$schema": "Transform配置示例",
  "$version": "1.0",
  "$description": "ElasticRelay Transform引擎配置",

  "transform_rules": [
    // 转换规则数组，每个规则针对特定的表
  ],

  "global_settings": {
    // 全局设置
  },

  "masking_templates": {
    // 可复用的脱敏模板
  }
}

转换规则基本属性

每个转换规则包含以下基本属性：

{
  "id": "rule-001",
  "name": "用户数据转换规则",
  "description": "用户表数据转换，包含脱敏和字段映射",
  "source_id": "mysql-main",
  "table_patterns": ["users", "user_profiles"],
  "enabled": true,
  "priority": 1,

  "field_mappings": [...],
  "field_configs": [...],
  "masking_rules": [...],
  "computed_fields": [...],
  "filters": [...]
}

属性	类型	必填	说明
`id`	string	✅	规则唯一标识
`name`	string	✅	规则名称
`source_id`	string	-	数据源 ID，空表示全局应用
`table_patterns`	string[]	-	表名匹配模式，支持通配符 `*`
`enabled`	boolean	✅	是否启用
`priority`	int	✅	优先级，数字越小优先级越高

表名匹配模式

{
  "table_patterns": [
    "users",           // 精确匹配 users 表
    "user_*",          // 前缀匹配: user_profiles, user_settings
    "*_log",           // 后缀匹配: access_log, error_log, audit_log
    "*_orders_*"       // 包含匹配
  ]
}

5. 字段映射：重命名与复制

字段映射用于在数据同步过程中重命名、复制或移动字段。

配置语法

{
  "field_mappings": [
    {
      "source_field": "原字段名",
      "target_field": "目标字段名",
      "action": "rename"
    }
  ]
}

操作类型

Action	描述	原字段	目标字段
`rename`	重命名字段	删除	创建
`copy`	复制字段	保留	创建
`move`	移动字段（同 rename）	删除	创建

实际应用示例

场景：统一字段命名规范

MySQL 表中使用下划线命名：user_name, created_at
Elasticsearch 索引需要驼峰命名：userName, createdAt

{
  "field_mappings": [
    {
      "source_field": "user_name",
      "target_field": "username",
      "action": "rename"
    },
    {
      "source_field": "created_at",
      "target_field": "createTime",
      "action": "copy"
    },
    {
      "source_field": "updated_at",
      "target_field": "updateTime",
      "action": "rename"
    }
  ]
}

转换效果：

输入 (MySQL):                      输出 (Elasticsearch):
{                                  {
  "user_name": "zhangsan",    →      "username": "zhangsan",
  "created_at": "2024-01-01",        "created_at": "2024-01-01",  ← copy保留原字段
  "updated_at": "2024-01-15"         "createTime": "2024-01-01",
}                                    "updateTime": "2024-01-15"
                                   }

嵌套字段支持

使用点号 . 访问嵌套字段（适用于 JSON 类型列）：

{
  "source_field": "user.address.city",
  "target_field": "city",
  "action": "copy"
}

6. 字段配置：类型转换与验证

字段配置用于类型转换、字段排除、默认值设置和数据验证。

配置语法

{
  "field_configs": [
    {
      "field": "字段名",
      "target_type": "目标类型",
      "required": false,
      "default_value": null,
      "null_strategy": "ignore",
      "exclude": false,
      "validation": {...}
    }
  ]
}

支持的数据类型

类型	描述	Elasticsearch 映射
`string`	字符串	text
`keyword`	关键字（不分词）	keyword
`text`	文本（分词）	text
`int`	32位整数	integer
`int64`	64位整数	long
`float`	32位浮点	float
`float64`	64位浮点	double
`bool`	布尔值	boolean
`date`	RFC3339 日期	date
`timestamp`	Unix 时间戳	long
`object`	JSON 对象	object

Null 值处理策略

策略	描述
`ignore`	忽略 null 值，保持原样
`default`	使用 `default_value` 替换 null
`error`	遇到 null 时报错
`remove`	从输出中移除该字段

实际应用示例

场景：用户表字段配置

{
  "field_configs": [
    {
      "field": "age",
      "target_type": "int",
      "required": false,
      "default_value": 0,
      "null_strategy": "default",
      "description": "年龄，null时默认为0"
    },
    {
      "field": "email",
      "target_type": "keyword",
      "required": true,
      "validation": {
        "pattern": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$",
        "error_message": "邮箱格式不正确"
      },
      "description": "邮箱，必填且需要格式验证"
    },
    {
      "field": "balance",
      "target_type": "float64",
      "null_strategy": "default",
      "default_value": 0.0,
      "description": "账户余额"
    },
    {
      "field": "is_vip",
      "target_type": "bool",
      "default_value": false,
      "description": "是否VIP会员"
    },
    {
      "field": "internal_notes",
      "exclude": true,
      "description": "内部备注，不同步到ES"
    },
    {
      "field": "debug_info",
      "exclude": true,
      "description": "调试信息，不同步到ES"
    }
  ]
}

转换效果：

输入 (MySQL):                       输出 (Elasticsearch):
{                                   {
  "age": "25",               →        "age": 25,              ← 字符串转整数
  "email": "test@example.com",        "email": "test@example.com",
  "balance": null,                    "balance": 0.0,          ← null转默认值
  "is_vip": 1,                        "is_vip": true,          ← 1转bool
  "internal_notes": "VIP客户",        // internal_notes 被排除
  "debug_info": "xxx"                 // debug_info 被排除
}                                   }

7. 数据脱敏：敏感信息保护

数据脱敏是 Transform 引擎最重要的功能之一，可以在数据同步过程中实时保护敏感信息。

脱敏策略类型

策略	描述	参数
`mask`	字符掩码	`prefix`（保留前N位）, `suffix`（保留后N位）, `char`（掩码字符）
`hash`	哈希加密	`algorithm`（sha256/md5）
`token`	令牌化	`prefix`（令牌前缀）
`regex`	正则替换	`pattern`（匹配模式）, `replace`（替换模式）

内置脱敏模板

ElasticRelay 提供了常用的脱敏模板，开箱即用：

模板	输入示例	输出示例
`phone`	`13812345678`	`138****5678`
`id_card`	`110101199001011234`	`1101**********1234`
`email`	`john.doe@example.com`	`jo***@example.com`
`bank_card`	`6222021234567890123`	`6222********0123`
`name`	`张三`	`张*`

方式一：使用预置模板

{
  "masking_rules": [
    {
      "field": "phone",
      "template": "phone",
      "description": "手机号脱敏: 138****5678"
    },
    {
      "field": "id_card",
      "template": "id_card",
      "description": "身份证脱敏: 1101**********1234"
    },
    {
      "field": "email",
      "template": "email",
      "description": "邮箱脱敏: jo***@example.com"
    },
    {
      "field": "bank_card",
      "template": "bank_card",
      "description": "银行卡脱敏: 6222********0123"
    }
  ]
}

方式二：自定义脱敏策略

{
  "masking_rules": [
    {
      "field": "password",
      "strategy": "hash",
      "params": {
        "algorithm": "sha256"
      },
      "description": "密码使用SHA256哈希"
    },
    {
      "field": "address",
      "strategy": "mask",
      "params": {
        "prefix": 6,
        "suffix": 0,
        "char": "*"
      },
      "description": "地址只保留前6个字符"
    },
    {
      "field": "user_ip",
      "strategy": "mask",
      "params": {
        "prefix": 0,
        "suffix": 0,
        "char": "*"
      },
      "description": "IP地址完全脱敏"
    }
  ]
}

定义可复用脱敏模板

在 masking_templates 中定义自己的脱敏模板：

{
  "masking_templates": {
    "phone": {
      "strategy": "mask",
      "params": {
        "prefix": 3,
        "suffix": 4,
        "char": "*"
      }
    },
    "id_card": {
      "strategy": "mask",
      "params": {
        "prefix": 4,
        "suffix": 4,
        "char": "*"
      }
    },
    "email": {
      "strategy": "regex",
      "params": {
        "pattern": "(.{2}).*(@.*)",
        "replace": "$1***$2"
      }
    },
    "custom_account": {
      "strategy": "mask",
      "params": {
        "prefix": 2,
        "suffix": 2,
        "char": "#"
      }
    }
  }
}

实际应用示例

完整的用户脱敏规则：

{
  "id": "user-data-transform",
  "name": "用户数据转换规则",
  "table_patterns": ["users", "user_profiles"],
  "enabled": true,
  "priority": 1,

  "masking_rules": [
    {
      "field": "phone",
      "template": "phone",
      "description": "手机号脱敏: 138****5678"
    },
    {
      "field": "id_card",
      "template": "id_card",
      "description": "身份证脱敏: 1101**********1234"
    },
    {
      "field": "email",
      "template": "email",
      "description": "邮箱脱敏: ab***@example.com"
    },
    {
      "field": "bank_card",
      "template": "bank_card",
      "description": "银行卡脱敏: 6222********1234"
    },
    {
      "field": "password",
      "strategy": "hash",
      "params": {
        "algorithm": "sha256"
      },
      "description": "密码SHA256哈希"
    },
    {
      "field": "address",
      "strategy": "mask",
      "params": {
        "prefix": 6,
        "suffix": 0,
        "char": "*"
      },
      "description": "地址部分脱敏"
    }
  ]
}

转换效果：

输入 (MySQL):                              输出 (Elasticsearch):
{                                          {
  "phone": "13812345678",            →       "phone": "138****5678",
  "id_card": "110101199001011234",           "id_card": "1101**********1234",
  "email": "zhangsan@example.com",           "email": "zh***@example.com",
  "bank_card": "6222021234567890123",        "bank_card": "6222********0123",
  "password": "mypassword123",               "password": "a665a45920422f9d...（SHA256）",
  "address": "北京市朝阳区建国路88号"           "address": "北京市朝阳区***"
}                                          }

8. 计算字段：动态派生新字段

计算字段功能允许你使用表达式动态计算新的字段值，非常适合创建业务派生指标。

配置语法

{
  "computed_fields": [
    {
      "field": "目标字段名",
      "expression": "计算表达式",
      "dependencies": ["依赖字段1", "依赖字段2"],
      "description": "描述"
    }
  ]
}

表达式语法

计算字段使用类 JavaScript 语法，使用 $ 符号访问数据字段：

$.field_name          // 直接字段访问
$.nested.field        // 嵌套字段访问
$.user.address.city   // 深层嵌套

支持的运算符

// 算术运算
$.price * $.quantity       // 乘法
$.total - $.discount       // 减法
$.value / 100              // 除法
$.count + 1                // 加法

// 比较运算
$.age > 18                 // 大于
$.status == "active"       // 等于
$.value != null            // 不等于

// 三元表达式（条件判断）
$.age < 18 ? "minor" : "adult"
$.score >= 60 ? "pass" : "fail"

// 多重条件
$.age < 18 ? 'minor' : ($.age < 60 ? 'adult' : 'senior')

内置函数

函数	描述	示例
`concat(...)`	字符串拼接	`concat($.first, ' ', $.last)`
`substr(s, start, len)`	截取子串	`substr($.name, 0, 5)`
`upper(s)`	转大写	`upper($.code)`
`lower(s)`	转小写	`lower($.email)`
`trim(s)`	去除空格	`trim($.text)`
`round(n, decimals)`	四舍五入	`round($.price, 2)`
`abs(n)`	绝对值	`abs($.diff)`
`floor(n)`	向下取整	`floor($.value)`
`ceil(n)`	向上取整	`ceil($.value)`
`now()`	当前时间戳	`now()`
`ifNull(v, default)`	空值处理	`ifNull($.name, "Unknown")`

实际应用示例

场景1：用户数据衍生

{
  "computed_fields": [
    {
      "field": "full_name",
      "expression": "concat($.last_name, $.first_name)",
      "dependencies": ["first_name", "last_name"],
      "description": "拼接中文姓名 (姓+名)"
    },
    {
      "field": "age_group",
      "expression": "$.age < 18 ? 'minor' : ($.age < 60 ? 'adult' : 'senior')",
      "dependencies": ["age"],
      "description": "年龄分组：未成年/成年/老年"
    },
    {
      "field": "processed_at",
      "expression": "now()",
      "description": "数据处理时间戳"
    },
    {
      "field": "display_balance",
      "expression": "round($.balance, 2)",
      "dependencies": ["balance"],
      "description": "格式化余额（保留2位小数）"
    }
  ]
}

场景2：订单数据衍生

{
  "computed_fields": [
    {
      "field": "total_amount",
      "expression": "$.price * $.quantity",
      "dependencies": ["price", "quantity"],
      "description": "计算订单总金额"
    },
    {
      "field": "is_completed",
      "expression": "$.status == 'delivered'",
      "dependencies": ["status"],
      "description": "是否已完成订单"
    },
    {
      "field": "discount_rate",
      "expression": "$.original_price > 0 ? round(($.original_price - $.final_price) / $.original_price * 100, 2) : 0",
      "dependencies": ["original_price", "final_price"],
      "description": "计算折扣率（百分比）"
    }
  ]
}

场景3：商品数据衍生

{
  "computed_fields": [
    {
      "field": "price_tier",
      "expression": "$.price < 1000 ? 'budget' : ($.price < 5000 ? 'mid-range' : 'premium')",
      "dependencies": ["price"],
      "description": "价格分级：平价/中端/高端"
    },
    {
      "field": "availability",
      "expression": "$.in_stock && $.stock_quantity > 0 ? 'available' : 'out_of_stock'",
      "dependencies": ["in_stock", "stock_quantity"],
      "description": "库存状态"
    },
    {
      "field": "stock_warning",
      "expression": "$.stock_quantity < 10 ? true : false",
      "dependencies": ["stock_quantity"],
      "description": "库存预警标记"
    }
  ]
}

转换效果：

输入 (MySQL):                       输出 (Elasticsearch):
{                                   {
  "first_name": "三",                 "first_name": "三",
  "last_name": "张",                  "last_name": "张",
  "age": 25,                          "age": 25,
  "balance": 1234.5678          →     "balance": 1234.5678,
}                                     "full_name": "张三",        ← 计算字段
                                      "age_group": "adult",       ← 计算字段
                                      "processed_at": "2024-01-15T10:30:00Z", ← 计算字段
                                      "display_balance": 1234.57  ← 计算字段
                                    }

9. 数据过滤：精确控制同步范围

数据过滤功能允许你根据条件决定哪些记录应该被同步，哪些应该被排除。

配置语法

{
  "filters": [
    {
      "field": "字段名",
      "operator": "操作符",
      "value": "比较值",
      "action": "动作",
      "description": "描述"
    }
  ]
}

操作符列表

操作符	描述	示例
`eq`	等于	`"status" eq "active"`
`ne`	不等于	`"status" ne "deleted"`
`gt`	大于	`"age" gt 18`
`gte`	大于等于	`"score" gte 60`
`lt`	小于	`"price" lt 100`
`lte`	小于等于	`"quantity" lte 10`
`in`	在列表中	`"type" in ["a", "b"]`
`nin`	不在列表中	`"status" nin ["deleted"]`
`regex`	正则匹配	`"email" regex ".*@example.com"`
`exists`	字段存在	`"email" exists true`

动作类型

动作	描述
`include`	条件匹配时保留记录，不匹配时排除
`exclude`	条件匹配时排除记录

实际应用示例

场景1：排除已删除和测试数据

{
  "filters": [
    {
      "field": "status",
      "operator": "ne",
      "value": "deleted",
      "action": "include",
      "description": "排除已删除记录"
    },
    {
      "field": "is_test",
      "operator": "ne",
      "value": 1,
      "action": "include",
      "description": "排除测试账号 (is_test=1表示测试账号)"
    }
  ]
}

场景2：只同步有效订单

{
  "filters": [
    {
      "field": "status",
      "operator": "nin",
      "value": ["cancelled", "refunded", "expired"],
      "action": "include",
      "description": "排除已取消、已退款、已过期订单"
    },
    {
      "field": "amount",
      "operator": "gt",
      "value": 0,
      "action": "include",
      "description": "排除金额为0的订单"
    }
  ]
}

场景3：只同步特定类型日志

{
  "filters": [
    {
      "field": "level",
      "operator": "in",
      "value": ["error", "warn", "info"],
      "action": "include",
      "description": "只同步 error/warn/info 级别日志"
    },
    {
      "field": "source",
      "operator": "ne",
      "value": "health_check",
      "action": "include",
      "description": "排除健康检查日志"
    }
  ]
}

过滤逻辑说明：

记录: {"status": "active", "is_test": 0}
→ status != "deleted" ✓, is_test != 1 ✓ → 保留 ✅

记录: {"status": "deleted", "is_test": 0}
→ status != "deleted" ✗ → 排除 ❌

记录: {"status": "active", "is_test": 1}
→ status != "deleted" ✓, is_test != 1 ✗ → 排除 ❌

10. 完整配置示例

下面是一个完整的 Transform 配置文件，展示了所有功能的综合应用：

`config/mysql_transform.json`

{
  "$schema": "Transform配置示例",
  "$version": "1.0",
  "$description": "ElasticRelay Transform引擎配置 - 完整示例",

  "transform_rules": [
    {
      "id": "user-data-transform",
      "name": "用户数据转换规则",
      "description": "用户表数据转换，包含脱敏和字段映射",
      "source_id": "mysql-main",
      "table_patterns": ["users", "user_profiles"],
      "enabled": true,
      "priority": 1,

      "field_mappings": [
        {
          "source_field": "user_name",
          "target_field": "username",
          "action": "rename"
        },
        {
          "source_field": "created_at",
          "target_field": "create_time",
          "action": "copy"
        },
        {
          "source_field": "updated_at",
          "target_field": "update_time",
          "action": "rename"
        }
      ],

      "field_configs": [
        {
          "field": "age",
          "target_type": "int",
          "required": false,
          "default_value": 0,
          "null_strategy": "default"
        },
        {
          "field": "email",
          "target_type": "keyword",
          "required": true,
          "validation": {
            "pattern": "^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$",
            "error_message": "Invalid email format"
          }
        },
        {
          "field": "balance",
          "target_type": "float64",
          "null_strategy": "default",
          "default_value": 0.0
        },
        {
          "field": "is_vip",
          "target_type": "bool",
          "default_value": false
        },
        {
          "field": "internal_notes",
          "exclude": true
        },
        {
          "field": "debug_info",
          "exclude": true
        }
      ],

      "masking_rules": [
        {
          "field": "phone",
          "template": "phone",
          "description": "手机号脱敏: 138****5678"
        },
        {
          "field": "id_card",
          "template": "id_card",
          "description": "身份证脱敏: 1101**********1234"
        },
        {
          "field": "email",
          "template": "email",
          "description": "邮箱脱敏: ab***@example.com"
        },
        {
          "field": "bank_card",
          "template": "bank_card",
          "description": "银行卡脱敏: 6222********1234"
        },
        {
          "field": "password",
          "strategy": "hash",
          "params": {
            "algorithm": "sha256"
          },
          "description": "密码SHA256哈希"
        },
        {
          "field": "address",
          "strategy": "mask",
          "params": {
            "prefix": 6,
            "suffix": 0,
            "char": "*"
          },
          "description": "地址部分脱敏"
        }
      ],

      "computed_fields": [
        {
          "field": "full_name",
          "expression": "concat($.last_name, $.first_name)",
          "dependencies": ["first_name", "last_name"],
          "description": "拼接中文姓名"
        },
        {
          "field": "age_group",
          "expression": "$.age < 18 ? 'minor' : ($.age < 60 ? 'adult' : 'senior')",
          "dependencies": ["age"],
          "description": "年龄分组"
        },
        {
          "field": "processed_at",
          "expression": "now()",
          "description": "处理时间戳"
        },
        {
          "field": "display_balance",
          "expression": "round($.balance, 2)",
          "dependencies": ["balance"],
          "description": "格式化余额"
        }
      ],

      "filters": [
        {
          "field": "status",
          "operator": "ne",
          "value": "deleted",
          "action": "include",
          "description": "排除已删除记录"
        },
        {
          "field": "is_test",
          "operator": "ne",
          "value": 1,
          "action": "include",
          "description": "排除测试账号"
        }
      ]
    },

    {
      "id": "order-data-transform",
      "name": "订单数据转换规则",
      "description": "订单表数据转换",
      "source_id": "mysql-main",
      "table_patterns": ["orders", "order_items"],
      "enabled": true,
      "priority": 2,

      "field_mappings": [
        {
          "source_field": "order_no",
          "target_field": "order_number",
          "action": "copy"
        }
      ],

      "field_configs": [
        {
          "field": "amount",
          "target_type": "float64",
          "required": true
        },
        {
          "field": "quantity",
          "target_type": "int",
          "default_value": 1
        },
        {
          "field": "order_date",
          "target_type": "date"
        },
        {
          "field": "status",
          "target_type": "keyword"
        },
        {
          "field": "shipping_address",
          "target_type": "text"
        }
      ],

      "masking_rules": [
        {
          "field": "shipping_address",
          "strategy": "mask",
          "params": {
            "prefix": 6,
            "suffix": 0,
            "char": "*"
          },
          "description": "收货地址部分脱敏"
        }
      ],

      "computed_fields": [
        {
          "field": "total_amount",
          "expression": "$.amount * $.quantity",
          "dependencies": ["amount", "quantity"],
          "description": "计算总金额"
        },
        {
          "field": "is_completed",
          "expression": "$.status == 'delivered'",
          "dependencies": ["status"],
          "description": "是否已完成"
        },
        {
          "field": "processed_at",
          "expression": "now()",
          "description": "处理时间戳"
        }
      ],

      "filters": [
        {
          "field": "status",
          "operator": "nin",
          "value": ["cancelled", "refunded"],
          "action": "include",
          "description": "排除已取消和已退款订单"
        }
      ]
    },

    {
      "id": "log-data-transform",
      "name": "日志数据转换规则",
      "description": "日志表数据转换，简化处理",
      "source_id": "mysql-main",
      "table_patterns": ["logs", "audit_logs", "*_log"],
      "enabled": true,
      "priority": 10,

      "field_configs": [
        {
          "field": "timestamp",
          "target_type": "date"
        },
        {
          "field": "level",
          "target_type": "keyword"
        },
        {
          "field": "message",
          "target_type": "text"
        },
        {
          "field": "request_body",
          "exclude": true
        },
        {
          "field": "response_body",
          "exclude": true
        }
      ],

      "masking_rules": [
        {
          "field": "user_ip",
          "strategy": "mask",
          "params": {
            "prefix": 0,
            "suffix": 0,
            "char": "*"
          },
          "description": "IP地址完全脱敏"
        }
      ]
    }
  ],

  "global_settings": {
    "default_null_strategy": "ignore",
    "enable_validation": true,
    "enable_computed_fields": true,
    "enable_masking": true,
    "max_expression_timeout_ms": 100,
    "cache_compiled_rules": true
  },

  "masking_templates": {
    "phone": {
      "strategy": "mask",
      "params": {
        "prefix": 3,
        "suffix": 4,
        "char": "*"
      }
    },
    "id_card": {
      "strategy": "mask",
      "params": {
        "prefix": 4,
        "suffix": 4,
        "char": "*"
      }
    },
    "email": {
      "strategy": "regex",
      "params": {
        "pattern": "(.{2}).*(@.*)",
        "replace": "$1***$2"
      }
    },
    "bank_card": {
      "strategy": "mask",
      "params": {
        "prefix": 4,
        "suffix": 4,
        "char": "*"
      }
    },
    "name": {
      "strategy": "mask",
      "params": {
        "prefix": 1,
        "suffix": 0,
        "char": "*"
      }
    }
  }
}

11. 启动与验证

11.1 配置启动脚本

编辑 start.sh：

#!/bin/bash

# ElasticRelay 启动脚本

# 数据源配置文件
CONFIG_FILE="./config/mysql_config.json"

# Transform 配置文件（启用数据转换）
TRANSFORM_CONFIG="./config/mysql_transform.json"

# 如需禁用 Transform，注释上一行或设置为空
# TRANSFORM_CONFIG=""

# 启动服务
./bin/elasticrelay \
  -config ${CONFIG_FILE} \
  -transform-config ${TRANSFORM_CONFIG} \
  -port 50051

11.2 命令行启动

# 添加执行权限
chmod +x start.sh

# 启动服务
./start.sh

# 或直接使用命令行
./bin/elasticrelay \
  -config ./config/mysql_config.json \
  -transform-config ./config/mysql_transform.json \
  -port 50051

11.3 验证同步效果

1. 检查服务状态

# 查看服务日志
tail -f logs/elasticrelay.log

# 检查 gRPC 端口
netstat -an | grep 50051

# 检查 metrics 端口
curl http://localhost:8080/metrics

2. 在 MySQL 中插入测试数据

-- 插入用户数据
INSERT INTO users (
  user_name, first_name, last_name, age, 
  phone, id_card, email, bank_card, 
  password, address, balance, is_vip, 
  status, is_test
) VALUES (
  'zhangsan', '三', '张', 25,
  '13812345678', '110101199001011234', 'zhangsan@example.com', '6222021234567890123',
  'password123', '北京市朝阳区建国路88号', 10000.50, 1,
  'active', 0
);

3. 在 Elasticsearch 中查询数据

# 查询同步的数据
curl -u elastic:password \
  'http://localhost:9200/elasticrelay_mysql_users/_search?pretty'

预期输出示例：

{
  "hits": {
    "hits": [
      {
        "_source": {
          "username": "zhangsan",
          "first_name": "三",
          "last_name": "张",
          "age": 25,
          "phone": "138****5678",
          "id_card": "1101**********1234",
          "email": "zh***@example.com",
          "bank_card": "6222********0123",
          "password": "5e884898da28047d9...（SHA256哈希）",
          "address": "北京市朝阳区***",
          "balance": 10000.50,
          "is_vip": true,
          "status": "active",
          "create_time": "2024-01-15T10:30:00Z",
          "update_time": "2024-01-15T10:30:00Z",
          "full_name": "张三",
          "age_group": "adult",
          "processed_at": "2024-01-15T10:30:05Z",
          "display_balance": 10000.50
        }
      }
    ]
  }
}

11.4 验证数据更新

-- 更新用户数据
UPDATE users SET age = 65 WHERE user_name = 'zhangsan';

查询 ES 确认 age_group 自动更新为 senior：

curl -u elastic:password \
  'http://localhost:9200/elasticrelay_mysql_users/_doc/1?pretty'

12. 性能调优与最佳实践

12.1 Transform 引擎性能数据

Transform 引擎经过深度优化，具有出色的性能表现：

操作	性能	内存占用
完整转换流水线	~800,000 ops/sec	1,601 B/op
字段映射	~4,500,000 ops/sec	416 B/op
类型转换	~22,000,000 ops/sec	16 B/op
过滤检查	~5,000,000 ops/sec	~200 B/op
数据脱敏	~1,000,000 ops/sec	~500 B/op

🚀 性能超越设计目标 80 倍！（设计目标: 10,000 ops/sec）

12.2 配置优化建议

1. 合理设置规则优先级

{
  "transform_rules": [
    { "id": "users", "priority": 1 },      // 高频表，高优先级
    { "id": "orders", "priority": 2 },
    { "id": "products", "priority": 5 },
    { "id": "logs", "priority": 10 }       // 低频表，低优先级
  ]
}

2. 启用规则缓存

{
  "global_settings": {
    "cache_compiled_rules": true
  }
}

3. 只配置必要的规则

不需要转换的表，不要添加规则
只配置需要转换的字段
避免过于复杂的表达式

4. 控制表达式复杂度

// ✅ 推荐：简单表达式
{
  "expression": "$.age < 18 ? 'minor' : 'adult'"
}

// ⚠️ 避免：过于复杂的嵌套
{
  "expression": "$.a ? ($.b ? ($.c ? 'x' : 'y') : 'z') : ($.d ? 'p' : 'q')"
}

5. 合理设置批量大小

{
  "jobs": [{
    "options": {
      "batch_size": 1000  // 根据数据量和网络情况调整
    }
  }]
}

12.3 监控指标

ElasticRelay 提供丰富的监控指标，通过 Prometheus 端点暴露：

curl http://localhost:8080/metrics

# 关键指标：
# elasticrelay_transform_processed_total    - 转换处理总数
# elasticrelay_transform_errors_total       - 转换错误总数
# elasticrelay_transform_latency_seconds    - 转换延迟
# elasticrelay_binlog_position              - Binlog 位置
# elasticrelay_sync_lag_seconds             - 同步延迟

13. 常见问题排查

Q1: Transform 规则不生效

排查步骤：

检查 enabled 是否为 true
确认 table_patterns 匹配目标表
验证 source_id 与数据源配置一致
查看日志确认规则是否正确加载

grep "transform" logs/elasticrelay.log

Q2: 脱敏后数据格式不对

排查步骤：

检查脱敏模板的 prefix、suffix 参数
确认字段值长度是否足够
使用 strategy: mask 时确保 char 参数正确

Q3: 计算字段表达式报错

排查步骤：

检查字段名是否正确（使用 $.fieldName）
确认依赖字段存在且有值
测试简单表达式，逐步增加复杂度

Q4: Binlog 同步延迟过大

排查步骤：

检查网络连接状态
增大 batch_size 配置
检查 Elasticsearch 写入性能
查看监控指标定位瓶颈

Q5: 如何热更新配置？

当前版本需要重启服务加载新配置。建议：

在测试环境验证配置
使用滚动重启减少影响
关注后续版本的热重载支持

Q6: 多个规则匹配同一表时如何处理？

按 priority 顺序依次应用所有匹配的规则，数字越小优先级越高。

总结

通过本教程，你已经掌握了 ElasticRelay Transform 引擎的核心功能：

字段映射 - 灵活重命名和复制字段
类型转换 - 自动适配 MySQL 到 Elasticsearch 的类型
数据脱敏 - 保护敏感信息，满足合规要求
计算字段 - 动态派生业务指标
数据过滤 - 精确控制同步范围

Transform 引擎让你在实时同步 MySQL 数据的同时，完成复杂的数据治理任务，是构建现代数据架构的强大工具。

所有评论(0)

查看更多评论

linjifan

@linjifan

已为社区贡献1条内容

ElasticRelay 实战：MySQL 到 Elasticsearch 实时同步与数据转换脱敏保姆级教程

linjifan

前言

目录

1. ElasticRelay 简介

Transform 引擎是什么？

Transform 处理流程

2. 环境准备

2.1 MySQL 配置要求

2.2 Elasticsearch 配置要求

2.3 ElasticRelay 安装

3. MySQL CDC 配置详解

配置项详解

4. Transform 引擎核心概念

配置文件基本结构

转换规则基本属性

表名匹配模式

5. 字段映射：重命名与复制

配置语法

操作类型

实际应用示例

嵌套字段支持

6. 字段配置：类型转换与验证

配置语法

支持的数据类型

Null 值处理策略

实际应用示例

7. 数据脱敏：敏感信息保护

脱敏策略类型

内置脱敏模板

方式一：使用预置模板

方式二：自定义脱敏策略

定义可复用脱敏模板

实际应用示例

8. 计算字段：动态派生新字段

配置语法

表达式语法

支持的运算符

内置函数

实际应用示例

9. 数据过滤：精确控制同步范围

配置语法

操作符列表

动作类型

实际应用示例

10. 完整配置示例

config/mysql_transform.json

11. 启动与验证

11.1 配置启动脚本

11.2 命令行启动

11.3 验证同步效果

11.4 验证数据更新

12. 性能调优与最佳实践

12.1 Transform 引擎性能数据

12.2 配置优化建议

12.3 监控指标

13. 常见问题排查

Q1: Transform 规则不生效

Q2: 脱敏后数据格式不对

Q3: 计算字段表达式报错

Q4: Binlog 同步延迟过大

Q5: 如何热更新配置？

Q6: 多个规则匹配同一表时如何处理？

总结

相关链接

所有评论(0)

温馨提示：您尚未绑定手机号

linjifan

`config/mysql_transform.json`