budgetkey/budgetkey-data-pipelines

budgetkey-data-pipelines

![Build Status]([***]

Budget Key数据处理管道

概述和主要用途

Budget Key项目的核心是其丰富、最新的高质量数据集。该数据从20多个不同数据源收集，经过清洗、标准化、验证、合并和分析，形成以色列最全面的财政数据存储库。

为获取这些数据，系统包含大量下载器和爬虫，从***出版物和其他网站获取数据。获取的数据经过处理和合并后，最终保存到磁盘（方便用户下载原始数据）、加载到关系数据库（供分析师进行深度查询）以及推送到键值存储（Elasticsearch），为主要网站（obudget.org）提供数据支持。

实现这一过程的框架是datapackage-pipelines，该框架允许编写简单的“管道”，每个管道由一组预定义处理步骤组成。这些管道无需编码，而是通过一组YAML文件定义。大多数管道使用通用构建块，部分使用自定义处理器（主要是针对特殊数据源的自定义爬虫）。

要查看各管道的当前处理状态，可访问仪表盘。

核心功能和特性

• 多数据源处理：支持从20多个***和公共数据源收集数据
• 声明式管道定义：通过YAML文件定义数据处理流程，无需编码
• 多样化数据处理：包含数据清洗、标准化、验证、合并和分析功能
• 多目标存储：数据可保存到磁盘、关系数据库和Elasticsearch
• 状态监控：通过仪表盘实时查看各管道处理状态
• 可扩展框架：支持通用构建块和自定义处理器扩展

使用场景和适用范围

• 财政数据分析：为分析师提供结构化财政数据，支持深度查询
• 数据公开：提供原始数据下载，促进财政数据透明化
• 网站数据服务：为obudget.org网站提供Elasticsearch支持的高效数据查询
• 数据研究：支持学术和政策研究，提供标准化的以色列财政数据集
• 数据管道开发：作为datapackage-pipelines框架的应用示例，供开发者参考

目录结构

所有管道定义位于datapackage_pipelines_budgetkey_data_pipeline/pipelines目录下，主要结构如下：

• budget/
  • national/
    • original：获取国家预算的管道
    • processed：处理和分析国家预算的管道
    • changes/
      • original：获取国家预算变更信息的管道
      • processed：处理和分析预算变更（检测交易等）的管道
      • explanations：获取和提取国家预算变更说明文档文本的管道
• entities/
  • associations：获取非***组织（NGO）信息的管道
  • companies：获取公司信息的管道
  • ottoman：获取奥斯曼协会信息的管道
  • special：获取其他实体信息的管道
• procurement/
  • spending：获取和处理***支出报告的管道
  • tenders：获取***招标流程数据的管道
• supports/：获取***支持及其相关流程数据的管道

注：如需了解不同类型***支出的区别，可阅读这篇博客文章。

开发新管道

1. 阅读datapackage-pipelines框架https://github.com/frictionlessdata/datapackage-pipelines
2. 按照下文“快速开始”部分的说明安装datapackage-pipelines-budgetkey包
3. 确定变更位置：
   • 是否与现有管道相关？
   • 是否为全新功能？
4. 参考现有类似管道，了解通用构建块和自定义处理器的使用方法

详细使用方法

安装准备

系统依赖安装

bash
sudo apt-get install build-essential python3-dev libxml2-dev libxslt1-dev libleveldb-dev

Python 3.6+安装

推荐使用https://github.com/pyenv/pyenv%E7%AE%A1%E7%90%86Python%E7%89%88%E6%9C%AC%E3%80%82

Ubuntu系统：

bash
sudo apt-get install git python-pip make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev
sudo pip install virtualenvwrapper

git clone https://github.com/yyuu/pyenv.git ~/.pyenv
git clone https://github.com/yyuu/pyenv-virtualenvwrapper.git ~/.pyenv/plugins/pyenv-virtualenvwrapper

echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.bashrc
echo 'export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.bashrc
echo 'eval "$(pyenv init -)"' >> ~/.bashrc
echo 'pyenv virtualenvwrapper' >> ~/.bashrc

exec $SHELL

OSX系统：

bash
brew install pyenv
echo 'eval "$(pyenv init -)"' >> ~/.bash_profile

安装完成后设置Python版本：

bash
pyenv install 3.6.1
pyenv global 3.6.1

包安装

bash
python --version  # 确保输出Python 3.6.0+
sudo mkdir -p /var/datapackages && sudo chown $USER /var/datapackages/
make install
budgetkey-dpp  # 验证安装，应显示可用管道列表

运行管道

bash
budgetkey-dpp run ./entities/companies/registrar/registry

运行后将创建以下文件：

• /var/datapackages：数据以datapackage格式保存
• datapackage_pipelines_budgetkey/.data.db：数据存储到数据库（可通过DPP_DB_ENGINE环境变量设置自定义数据库，格式为SQLAlchemy连接URL）
• datapackage_pipelines_budgetkey/pipelines/.dpp.db：管道元数据和运行状态

编写测试

单元测试

bash
make test

运行特定测试/修改测试参数

bash
# 运行特定测试文件
tox tests/tenders/test_fixtures.py

# 直接使用py.test（需先安装pytest）
pip install pytest
py.test tests/tenders/test_fixtures.py -svk test_tenders_fixtures_publishers

使用Docker Compose

Docker Compose可用于运行包含所有必要服务的完整环境，类似生产环境。

安装

1. 安装Docker和Docker Compose（参考Docker官方文档）
2. （可选）从当前目录构建镜像：docker-compose build pipelines
3. 后台启动最小化必要服务：

   bash
   docker-compose up -d redis db pipelines
   

可用服务端点

• 管道仪表盘：http://localhost:5000/（默认不运行工作节点）
• 数据库：postgresql://postgres:123456@localhost:15432/postgres

在容器内运行命令

bash
docker-compose exec pipelines sh -c "budgetkey-dpp"

启动Elasticsearch和Kibana

bash
docker-compose up -d elasticsearch kibana

从本地连接容器服务

bash
source .env.example
dpp

加载数据到Elasticsearch

以下步骤将datapackage加载到Elasticsearch，供Kibana探索数据（注意：会删除本地Docker Compose卷，确保无重要数据）：

bash
# 启动基础服务
docker-compose down -v && docker-compose pull elasticsearch db && docker-compose up -d elasticsearch db

# 设置环境变量
export DPP_DB_ENGINE="postgresql://postgres:123456@localhost:15432/postgres"
export DPP_ELASTICSEARCH="localhost:19200"

# 加载数据（默认加载每个管道前100行，可修改ES_LIMIT_ROWS或移除）
for doctype in `budgetkey-dpp | grep .budgetkey/elasticsearch/index_ | cut -d"_" -f2 - | cut -d" " -f1 -`; do
    echo " > 加载 ${doctype}"
    ES_LOAD_FROM_URL=1 ES_LIMIT_ROWS=100 budgetkey-dpp run ./budgetkey/elasticsearch/index_$doctype
done

启动Kibana探索数据：

bash
docker-compose up -d kibana

Kibana可通过http://localhost:15601/访问（启动可能需要几分钟），索引名称为budgetkey。