数据同步在数据平台中扮演着至关重要的角色。它们用于从各种来源提取数据，对数据进行处理和分析，然后将数据存储到目标位置。数据同步可以用于各种目的，包括预测分析、商业智能、数据集成和数据迁移。

在本文中，我们将简单回顾数据同步的一些主要工作。然后，了解如何使用开源Python ETL 工具Singer创建数据同步作业。

数据同步作业回顾

正如输水管道将水从源头输送到目的地一样，我们通过数据同步工作将数据从一个源系统输送到目的地系统。数据同步几个场景：

1. 异构数据存储之间的数据同步。例如，将 Oracle 数据库实例的数据同步到 MySQL 数据库实例。

2. 在数据清理、重构、汇总等之后，将数据从暂存区域转移到生产系统。例如，从各个网站抓取的数据可以收集在暂存区域中，然后以结构化方式适合消费，然后传输到数据库。

3. 从中央数据源跨各个子系统分发数据。例如：组织在 Excel Sheets 中收集的各种产品的调查数据可以被划分并分发到相应的产品团队进行处理。

数据同步作业的主要关注点

1. 数据频率：目标系统需要数据的速度，即定期小批量或实时获取数据。同步工作应足以维持目标系统所需的数据传输频率。

2. 弹性：数据同步的容错性和弹性至关重要，即当同步作业因突然的数据负载或代码漏洞崩溃时，不应造成数据丢失。

3. 可扩展性：用于开发数据同步作业的工具和技术必须具备重新配置的能力，以便在数据负载增加时扩展到更多硬件节点。

Singer工具介绍

Singer 的主要目的是成为“用于编写移动数据脚本的开源规范”。它涉及使用标准化的脚本进行数据提取和抽取，这些脚本可以根据需求与各种源/目标混合和匹配。

Singer核心功能

1. Tap-想从哪个数据源取数据？在 Singer 里叫 Tap，官网提供的现成 Tap 随意挑选，不满意还能自己写一个！

2. Target-数据目标，即负责从 Tap 拉取数据的对象，在 Singer 中被称为 Target。与 Tap 一样，你也可以直接使用 Singer 官网提供的现成 Target，或是根据需求创建自定义的 Target。

3. 数据交换格式-Singer 采用 JSON（JavaScript 对象表示法）作为数据交换格式，实现了对数据源和目标的无差别支持。

4. Tap 和 Target 可以轻松地使用 Unix 管道符（|）连接，无需任何守护进程或复杂的插件。例如：

tap | target

5. Singer 支持增量提取，通过在调用之间维护状态来实现。也就是说，可以在调用之间将时间戳存储在 JSON 文件中，以记录目标上次消费数据的时间点。

Singer工具的组件

我们可以根据我们的需求为 Singer 创建自定义Taps/Sinks，或者仅安装并使用 Singer 网站上已有的Taps/Sinks。

使用Singer构建自定义数据同步作业

在此demo中，我们将创建自己的 Tap 以从 REST API 获取员工记录。

1. 创建并激活 virtualenv

python3 -m venv ~/.virtualenvs/Singer.io_rest_tapsource ~/.virtualenvs/Singer.io_rest_tap/bin/activate

2. 安装 Singer python 库

pip install singer-python

3. 在编辑器中创建一个名为 tap_emp_api.py 的新文件并添加以下代码：

import singerimport urllib.request  
import urllib.requestimport json
#REST API传递的样本数据的JSON格式
schema = {‘properties’: {‘id’: {‘type’: ‘string’},‘employee_name’: {‘type’: ‘string’},‘employee_salary’: {‘type’: ‘string’},‘employee_age’: {‘type’: ‘string’},‘profile_image’: {‘type’: ‘string’}}}
#提交http请求并解析响应
with urllib.request.urlopen(‘http://dummy.restapiexample.com/api/v1/employees') as response:
emp_data = json.loads(response.read().decode(‘utf-8’))
# 接下来我们调用singer.write_schema
singer.write_schema(‘employees’, schema, ‘id’)
#然后我们调用singer.write_records将记录写入该流
singer.write_records(‘employees’, records=emp_data[“data”])

设置写入目标

Singer 网站上已经提供了 PostgreSQL 的Target，因此我们将设置一个 virtualenv 来安装它。

1. 创建并激活 virtualenv

python3 -m venv ~/.virtualenvs/Singer.io_postgres_targetsource ~/.virtualenvs/Singer.io_ postgres_target /bin/activate

2. 安装 Singer python 库

 pip install singer-target-postgres

3. 接下来需要创建配置文件，用来连接到postgres数据库，如下所示：

{
  “postgres_host”: “localhost”,
 “postgres_port”: 5432,
 “postgres_database”: “singer_demo”,
 “postgres_username”: “postgres”,
 “postgres_password”: “postgres”,
 “postgres_schema”: “public”
}

使用Singer数据同步作业

Tap 和 Target 已经设置完毕： 这意味着已经完成了 Tap 和 Target 的配置，它们可以连接到数据源和目标系统。

python ~/.virtualenvs/Singer.io_rest_tap/bin/tap_emp_api.py | ~/.virtualenvs/Singer.io_postgres_target/bin/target-postgres — config database_config.json

完成！所有从 API 获取的记录都会按照我们为 Tap 设置的 database_config 文件和Schema插入到表中。是不是很简单？！

随着数据量的不断增长，数据同步将变得越来越重要。数据同步可以帮助组织有效地处理和分析数据，从中提取价值。

除了 Singer，你还可以了解其他数据同步编排工具，例如功能强大的 Airflow 和 Luigi。