数据库同步 Elasticsearch 后数据不一致，怎么办？

news2024/12/23 5:50:52

1、实战线上问题

Q1：Logstash 同步 postgreSQL 到 Elasticsearch 数据不一致。

在使用 Logstash 从 pg 库中将一张表导入到 ES 中时，发现 ES 中的数据量和 PG 库中的这张表的数据量存在较大差距。如何快速比对哪些数据没有插入？导入过程中，Logstash 日志没有异常。PG 中这张表有 7600W。

Q2：mq 异步双写数据库、es 的方案中，如何保证数据库数据和 es 数据的一致性？

2、推荐解决方案之一——ID 比较法

如下示例，仅拿问题1举例验证，问题2原理一致。

2.1 方案探讨

要找出哪些数据没有插入到 Elasticsearch 中，可以采用以下方法：

确保 Logstash 配置文件中的 input 插件的 JDBC 驱动程序正确配置，以便从 PostgreSQL 数据库中提取所有数据。注意 statement 参数，确保它选择了所有需要的数据。
检查 Logstash 配置文件的 output 插件，确保正确配置了 Elasticsearch 的连接参数。同时，检查是否有过滤器在导入过程中过滤掉了部分数据。
在 Logstash 配置文件中添加一个 stdout 插件，将从 PostgreSQL 数据库中读取的数据记录到文件中。

例如，可以添加以下内容：

output {
  elasticsearch {
    ...Elasticsearch 配置...
  }
  stdout {
    codec => json_lines
    path => "/path/to/logstash_output.log"
  }
}

将 Logstash 输出文件与 PostgreSQL 数据库中的原始数据进行比较，以找出未导入的数据。可以使用 Python、Shell 脚本或其他编程语言编写一个简单的脚本来执行此操作。

如果 Logstash 输出文件中的记录数与 PostgreSQL 数据库中的记录数一致，但 Elasticsearch 中的记录数不一致，请检查 Elasticsearch 集群的健康状况和日志。确认集群是否在接收和索引数据时遇到问题。

如果问题仍然存在，尝试将批量操作的大小减小，以减轻 Elasticsearch 和 Logstash 的负担。可以通过在 Logstash 配置文件的 output 插件中设置 flush_size 和 idle_flush_time 参数来实现。

处理大量数据时，可能需要调整 Logstash 和 Elasticsearch 的性能和资源配置。根据硬件和网络条件，可能需要优化批量操作、JVM 设置、线程池大小等方面的设置。

2.2 比较脚本的实现

以下是一个简单的 Shell 脚本示例，用于比较 Logstash 输出文件（JSON 格式）和 PostgreSQL 数据库中的数据。该脚本将比较特定字段（如 id）以确定哪些数据可能未导入到 Elasticsearch。

首先，从 PostgreSQL 数据库中导出数据，将其保存为 CSV 文件：

COPY (SELECT id FROM your_table) TO '/path/to/postgres_data.csv' WITH

接下来，创建一个名为 compare.sh 的 Shell 脚本：

#!/bin/bash
# 将 JSON 文件中的 ID 提取到一个文件中
jq '.id' /path/to/logstash_output.log > logstash_ids.txt

# 删除 JSON 中的双引号
sed -i 's/"//g' logstash_ids.txt

# 对 Logstash 和 PostgreSQL 的 ID 文件进行排序
sort -n logstash_ids.txt > logstash_ids_sorted.txt
sort -n /path/to/postgres_data.csv > postgres_ids_sorted.txt

# 使用 comm 比较两个已排序的 ID 文件
comm -23 postgres_ids_sorted.txt logstash_ids_sorted.txt > missing_ids.txt

# 输出结果
echo "以下 ID 在 Logstash 输出文件中未找到："
cat missing_ids.txt

为脚本添加可执行权限并运行：

chmod +x compare.sh

./compare.sh

此脚本会比较 logstash_output.log 和 postgres_data.csv 文件中的 ID。如果发现缺失的 ID，它们将被保存在 missing_ids.txt 文件中，并输出到控制台。请注意，该脚本假设已经安装了 jq（一个命令行 JSON 处理器）。如果没有，请先安装 jq。

3、推荐方案二——Redis 加速对比

在这种情况下，可以使用 Redis 的集合数据类型来存储 PostgreSQL 数据库和 Logstash 输出文件中的 ID。接下来，可以使用 Redis 提供的集合操作来找到缺失的 ID。

以下是一个使用 Redis 实现加速比对的示例：

首先，从 PostgreSQL 数据库中导出数据，将其保存为 CSV 文件：

COPY (SELECT id FROM your_table) TO '/path/to/postgres_data.csv' WITH CSV HEADER;

安装并启动 Redis。

使用 Python 脚本将 ID 数据加载到 Redis：

import redis
import csv

# 连接到 Redis

r = redis.StrictRedis(host='localhost', port=6379, db=0)

# 从 PostgreSQL 导出的 CSV 文件中加载数据
with open('/path/to/postgres_data.csv', newline='') as csvfile:
    csv_reader = csv.reader(csvfile)
    next(csv_reader)  # 跳过表头
    for row in csv_reader:
        r.sadd('postgres_ids', row[0])

# 从 Logstash 输出文件中加载数据
with open('/path/to/logstash_output.log', newline='') as logstash_file:
    for line in logstash_file:
        id = line.split('"id":')[1].split(',')[0].strip()
        r.sadd('logstash_ids', id)

# 计算差集
missing_ids = r.sdiff('postgres_ids', 'logstash_ids')

# 输出缺失的 ID
print("以下 ID 在 Logstash 输出文件中未找到：")
for missing_id in missing_ids:
    print(missing_id)

这个 Python 脚本使用 Redis 集合数据类型存储 ID，然后计算它们之间的差集以找到缺失的 ID。需要先安装 Python 的 Redis 库。可以使用以下命令安装：