跳转到主要内容

Express Loader - 快速将数据加载到CKAN DataStore

项目描述

https://travis-ci.org/ckan/ckanext-xloader.svg?branch=master Latest Version Supported Python versions Development Status License

XLoader - ckanext-xloader

将CSV(及类似)数据加载到CKAN的数据存储中。作为DataPusher的替代品,它提供了十倍的速度和更强的健壮性(因此得名,来源于“Express Loader”

OpenGov Inc. 赞助了这一开发,旨在造福全球的开放数据基础设施。

与DataPusher的关键区别

加载速度

DataPusher - 解析CSV行,将其转换为检测到的列类型,将数据转换为JSON字符串,为每批行调用datastore_create,将数据重新格式化为INSERT语句字符串,然后传递给PostgreSQL。

XLoader - 使用COPY直接将CSV文件导入PostgreSQL。

测试中,XLoader比DataPusher快十倍以上。

健壮性

DataPusher - 失败的一个原因是将单元格转换为猜测的类型。列的类型是通过查看前几行的值来决定的。因此,如果一个列主要包含数字或日期,但后面出现了一个字符串(如“N/A”),那么这将在该点导致加载错误,使得它只加载到DataStore的一半。

XLoader - 在允许管理员将列转换为所需的类型之前(使用数据字典功能),将所有单元格作为文本加载。未来它可能进行自动检测和转换。

更简单的排队技术

DataPusher - 作业队列由定制的、复杂的ckan-service-provider完成,它存储作业在其自己的数据库中(默认为sqlite)。

XLoader - 作业队列由RQ完成,它更简单,由Redis支持,允许访问CKAN模型,并且是CKAN的默认队列技术(自CKAN 2.7起)。您还可以使用pdb轻松调试作业。作业结果默认存储在Sqlite中,对于生产环境,只需在配置中指定CKAN的数据库即可,它将存储在那里 - 很简单。

(另一个明显的候选者是Celery,但我们不需要它的重量级架构,并且其作业无法使用pdb调试。)

独立的Web服务器

DataPusher - 队列作业由一个独立的(Flask)Web应用完成,除了CKAN之外。这种设计是因为转换每行数据为JSON需要密集的处理。然而,这意味着代码更复杂,因为需要在服务之间通过http请求传递信息,用户需要设置和管理更多 - 另一个应用程序配置,另一个apache配置,独立的日志文件。

XLoader - 作业在同一个应用程序的worker进程中运行,与CKAN相同,因此可以直接访问CKAN配置、数据库和日志记录,从而避免了许多HTTP调用。这种简化是有意义的,因为xloader作业不需要做很多处理 - 主要是将CSV文件从磁盘流式传输到PostgreSQL。

注意事项 - 列类型

注意:使用XLoader,所有列都存储在DataStore数据库的'text'类型中(而DataPusher进行了一些基本的类型猜测 - 见“健壮性”上述)。然而,一旦资源被xloaded,管理员可以使用资源的“数据字典”标签(从CKAN 2.7开始)将这些类型更改为数值或日期戳并重新加载文件。当从DataPusher迁移到XLoader时,您可以使用migrate_types命令保留现有资源的类型。

有空间添加自动猜测列类型的功能 - 欢迎贡献。

要求

与核心CKAN版本的兼容性

CKAN版本

兼容性

2.3

不再测试,您必须安装ckanext-rq

2.4

不再测试,您必须安装ckanext-rq

2.5

不再测试,您必须安装ckanext-rq

2.6

不再测试,您必须安装ckanext-rq

2.7

2.8

2.9

是(Python2和Python3)

安装

要安装XLoader

  1. 激活您的CKAN虚拟环境,例如

    . /usr/lib/ckan/default/bin/activate
  2. 将 ckanext-xloader Python 包安装到您的虚拟环境中

    pip install ckanext-xloader
  3. 安装依赖项

    pip install -r https://raw.githubusercontent.com/ckan/ckanext-xloader/master/requirements.txt
    pip install -U requests[security]
  4. 如果您使用的是 2.8.x 版本之前的 CKAN,您需要在数据库中定义 populate_full_text_trigger

    sudo -u postgres psql datastore_default -f full_text_function.sql

    如果成功,它将打印

    CREATE FUNCTION
    ALTER FUNCTION

    注意:此假设您使用了数据库名称和用户名的默认值。如果有疑问,请检查您配置的 ckan.datastore.write_url。如果您没有数据库名称 datastore_default 和用户名 ckan_default,则在运行此操作之前调整 psql 选项和 full_text_function.sql

  5. xloader 添加到您的 CKAN 配置文件中的 ckan.plugins 设置(默认配置文件位于 /etc/ckan/default/production.ini)。

    如果列表中包含 datapusher,您也应该将其删除,以避免两者都尝试将资源加载到 DataStore 中。

    确保 datastore 也被列出,以启用 CKAN DataStore。

  6. 从 CKAN 2.10 开始,您需要设置 API 令牌才能执行针对服务器的作业

    ckanext.xloader.api_token = <your-CKAN-generated-API-Token>
  7. 如果是生产服务器,您可能希望将作业信息存储在比默认 sqlite 文件更健壮的数据库中。它可以通过添加此行到配置文件中,高兴地使用主 CKAN postgres 数据库,但使用与 sqlalchemy.url 相同的值

    ckanext.xloader.jobs_db.uri = postgresql://ckan_default:pass@localhost/ckan_default

    (当只是开发或测试时,可以跳过此步骤。)

  8. 重新启动 CKAN。例如,如果您已在 Ubuntu 上使用 Apache 部署 CKAN

    sudo service apache2 reload
  9. 运行工作进程。首先在命令行中测试它。如果您有 CKAN 版本 2.9 或更高版本

    ckan -c /etc/ckan/default/ckan.ini jobs worker

    否则

    paster --plugin=ckan jobs -c /etc/ckan/default/ckan.ini worker

    或者如果您有 CKAN 版本 2.6.x 或更低版本(并且因此使用 ckanext-rq)

    paster --plugin=ckanext-rq jobs -c /etc/ckan/default/ckan.ini worker

    通过提交 web 界面中的 CSV 或在其他 shell 中测试它是否可以正常加载 CSV

    paster --plugin=ckanext-xloader xloader submit <dataset-name> -c /etc/ckan/default/ckan.ini

    显然,在命令行上运行工作进程仅用于测试 - 对于生产服务,请参阅

    http://docs.ckan.org/en/ckan-2.7.0/maintaining/background-tasks.html#using-supervisor

    如果您有 CKAN 版本 2.6.x 或更低版本,则需要下载 supervisor-ckan-worker.conf 并调整 command 以引用 ckanext-rq。

配置设置

配置

# The connection string for the jobs database used by XLoader. The
# default of an sqlite file is fine for development. For production use a
# Postgresql database.
ckanext.xloader.jobs_db.uri = sqlite:////tmp/xloader_jobs.db

# The formats that are accepted. If the value of the resource.format is
# anything else then it won't be 'xloadered' to DataStore (and will therefore
# only be available to users in the form of the original download/link).
# Case insensitive.
# (optional, defaults are listed in plugin.py - DEFAULT_FORMATS).
ckanext.xloader.formats = csv application/csv xls application/vnd.ms-excel

# The maximum size of files to load into DataStore. In bytes. Default is 1 GB.
ckanext.xloader.max_content_length = 1000000000

# By default, xloader will first try to add tabular data to the DataStore
# with a direct PostgreSQL COPY. This is relatively fast, but does not
# guess column types. If this fails, xloader falls back to a method more
# like DataPusher's behaviour. This has the advantage that the column types
# are guessed. However it is more error prone and far slower.
# To always skip the direct PostgreSQL COPY and use type guessing, set
# this option to True.
ckanext.xloader.use_type_guessing = False

# Deprecated: use ckanext.xloader.use_type_guessing instead.
ckanext.xloader.just_load_with_messytables = False

# Whether ambiguous dates should be parsed day first. Defaults to False.
# If set to True, dates like '01.02.2022' will be parsed as day = 01,
# month = 02.
# NB: isoformat dates like '2022-01-02' will be parsed as YYYY-MM-DD, and
# this option will not override that.
# See https://dateutil.readthedocs.io/en/stable/parser.html#dateutil.parser.parse
# for more details.
ckanext.xloader.parse_dates_dayfirst = False

# Whether ambiguous dates should be parsed year first. Defaults to False.
# If set to True, dates like '01.02.03' will be parsed as year = 2001,
# month = 02, day = 03. See https://dateutil.readthedocs.io/en/stable/parser.html#dateutil.parser.parse
# for more details.
ckanext.xloader.parse_dates_yearfirst = False

# The maximum time for the loading of a resource before it is aborted.
# Give an amount in seconds. Default is 60 minutes
ckanext.xloader.job_timeout = 3600

# Ignore the file hash when submitting to the DataStore, if set to True
# resources are always submitted (if their format matches), if set to
# False (default), resources are only submitted if their hash has changed.
ckanext.xloader.ignore_hash = False

# When loading a file that is bigger than `max_content_length`, xloader can
# still try and load some of the file, which is useful to display a
# preview. Set this option to the desired number of lines/rows that it
# loads in this case.
# If the file-type is supported (CSV, TSV) an excerpt with the number of
# `max_excerpt_lines` lines will be submitted while the `max_content_length`
# is not exceeded.
# If set to 0 (default) files that exceed the `max_content_length` will
# not be loaded into the datastore.
ckanext.xloader.max_excerpt_lines = 100

# Requests verifies SSL certificates for HTTPS requests. Setting verify to
# False should only be enabled during local development or testing. Default
# to True.
ckanext.xloader.ssl_verify = True

# Uses a specific API token for the xloader_submit action instead of the
# apikey of the site_user
ckanext.xloader.api_token = ckan-provided-api-token

开发者安装

要为开发安装 XLoader,激活您的 CKAN 虚拟环境,并在您的本地 ckan 仓库上方的目录中

git clone https://github.com/ckan/ckanext-xloader.git
cd ckanext-xloader
python setup.py develop
pip install -r requirements.txt
pip install -r dev-requirements.txt

从 DataPusher 升级

要从 DataPusher 升级到 XLoader

  1. 如上所述安装 XLoader,包括运行 xloader 工作进程。

  2. (可选)对于已将数据推送到数据存储的现有数据集,冻结列类型(在数据字典中),这样 XLoader 就不会在下次 xload 时将它们改回字符串

    ckan -c /etc/ckan/default/ckan.ini migrate_types
  3. 如果您尚未更改配置文件中的启用插件 - 在 ckan.plugins 行中,将 datapusher 替换为 xloader

  4. (可选)如果您愿意,可以禁用直接加载并继续仅使用 tabulator - 更多信息请参阅配置选项的文档:ckanext.xloader.use_type_guessing

  5. 停止 datapusher 工作进程

    sudo a2dissite datapusher
  6. 重新启动 CKAN

    sudo service apache2 reload
    sudo service nginx reload

命令行界面

您可以使用命令行界面提交单个或多个资源进行 xload。

例如。

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader submit <dataset-name>
[pre-2.9] paster --plugin=ckanext-xloader xloader submit <dataset-name> -c /etc/ckan/default/ckan.ini

对于调试,您可以通过使用 -s 选项同步地 xload 它(这直接执行加载,而不是请求工作进程执行)

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader submit <dataset-name> -s
[pre-2.9] paster --plugin=ckanext-xloader xloader submit <dataset-name> -s -c /etc/ckan/default/ckan.ini

查看作业状态

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader status
[pre-2.9] paster --plugin=ckanext-xloader xloader status -c /etc/ckan/default/development.ini

将所有数据集的资源提交到 DataStore

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader submit all
[pre-2.9] paster --plugin=ckanext-xloader xloader submit all -c /etc/ckan/default/ckan.ini

重新提交已提交到 DataStore 的所有资源(忽略任何未存储在 DataStore 中的资源,例如,因为它们不是表格数据)

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader submit all-existing
[pre-2.9] paster --plugin=ckanext-xloader xloader submit all-existing -c /etc/ckan/default/ckan.ini

XLoader CLI 命令的完整列表:

[2.9] ckan -c /etc/ckan/default/ckan.ini xloader --help
[pre-2.9] paster --plugin=ckanext-xloader xloader --help

作业和工作进程

管理作业的主要文档:<https://docs.ckan.org/en/latest/maintaining/background-tasks.html#managing-background-jobs>

运行和管理工作者进程的主要文档在这里:https://docs.ckan.org/en/latest/maintaining/background-tasks.html#running-background-jobs

有用的命令

清除(删除)所有挂起的作业

CKAN 2.9, Python 3 ckan -c /etc/ckan/default/ckan.ini jobs clear [QUEUES]
CKAN <2.9, Python 2 paster --plugin=ckanext-xloader xloader jobs clear [QUEUES] -c /etc/ckan/default/development.ini

如果遇到工作者进程的问题,重新启动它可能会有帮助

sudo supervisorctl restart ckan-worker:*

故障排除

KeyError: “Action ‘datastore_search’ not found”

您需要在CKAN配置中启用datastore插件。请参阅上面的“安装”部分以执行此操作,然后重新启动工作者。

ProgrammingError: (ProgrammingError) relation “_table_metadata” does not exist

您的DataStore权限尚未设置 - 请参阅:<https://docs.ckan.org/en/latest/maintaining/datastore.html#set-permissions>

在编辑包时,所有现有资源都会通过xloader重新加载

此行为在问题75中有记录,并且与CKAN中的错误有关,该错误已在2.6.9、2.7.7、2.8.4和2.9.0+版本中修复。

运行测试

第一次,您的测试数据存储数据库需要应用触发器

sudo -u postgres psql datastore_test -f full_text_function.sql

要运行测试,请执行

nosetests --nologcapture --with-pylons=test.ini

要运行测试并生成覆盖率报告,首先确保您已安装coverage在您的虚拟环境中(pip install coverage),然后运行

nosetests --nologcapture --with-pylons=test.ini --with-coverage --cover-package=ckanext.xloader --cover-inclusive --cover-erase --cover-tests

发布XLoader的新版本

XLoader可在PyPI上作为https://pypi.ac.cn/project/ckanext-xloader获得。

要将新版本发布到PyPI,请按照以下步骤操作

  1. 更新setup.py文件中的版本号。有关如何选择版本号的说明,请参阅PEP 440

  2. 更新CHANGELOG。

  3. 确保您有必要的最新版本包

    pip install --upgrade setuptools wheel twine
  4. 创建新版本的源和二进制发行版

    python setup.py sdist bdist_wheel && twine check dist/*

    修复您遇到的任何错误。

  5. 将源发行版上传到PyPI

    twine upload dist/*
  6. 提交任何挂起的更改

    git commit -a
    git push
  7. 在GitHub上将项目的最新版本标记为来自setup.py文件的版本号。例如,如果setup.py中的版本号为0.0.1,则执行

    git tag 0.0.1
    git push --tags

项目详情


下载文件

下载适合您平台的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。

源发行版

ckanext-xloader-0.12.2.tar.gz (77.0 kB 查看哈希值)

上传时间

支持者

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF 赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误日志 StatusPage StatusPage 状态页面