Express Loader - 快速将数据加载到CKAN DataStore
项目描述
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
激活您的CKAN虚拟环境,例如
. /usr/lib/ckan/default/bin/activate
将 ckanext-xloader Python 包安装到您的虚拟环境中
pip install ckanext-xloader
安装依赖项
pip install -r https://raw.githubusercontent.com/ckan/ckanext-xloader/master/requirements.txt pip install -U requests[security]
如果您使用的是 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
将 xloader 添加到您的 CKAN 配置文件中的 ckan.plugins 设置(默认配置文件位于 /etc/ckan/default/production.ini)。
如果列表中包含 datapusher,您也应该将其删除,以避免两者都尝试将资源加载到 DataStore 中。
确保 datastore 也被列出,以启用 CKAN DataStore。
从 CKAN 2.10 开始,您需要设置 API 令牌才能执行针对服务器的作业
ckanext.xloader.api_token = <your-CKAN-generated-API-Token>
如果是生产服务器,您可能希望将作业信息存储在比默认 sqlite 文件更健壮的数据库中。它可以通过添加此行到配置文件中,高兴地使用主 CKAN postgres 数据库,但使用与 sqlalchemy.url 相同的值
ckanext.xloader.jobs_db.uri = postgresql://ckan_default:pass@localhost/ckan_default
(当只是开发或测试时,可以跳过此步骤。)
重新启动 CKAN。例如,如果您已在 Ubuntu 上使用 Apache 部署 CKAN
sudo service apache2 reload
运行工作进程。首先在命令行中测试它。如果您有 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
如上所述安装 XLoader,包括运行 xloader 工作进程。
(可选)对于已将数据推送到数据存储的现有数据集,冻结列类型(在数据字典中),这样 XLoader 就不会在下次 xload 时将它们改回字符串
ckan -c /etc/ckan/default/ckan.ini migrate_types
如果您尚未更改配置文件中的启用插件 - 在 ckan.plugins 行中,将 datapusher 替换为 xloader。
(可选)如果您愿意,可以禁用直接加载并继续仅使用 tabulator - 更多信息请参阅配置选项的文档:ckanext.xloader.use_type_guessing
停止 datapusher 工作进程
sudo a2dissite datapusher
重新启动 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,请按照以下步骤操作
更新setup.py文件中的版本号。有关如何选择版本号的说明,请参阅PEP 440。
更新CHANGELOG。
确保您有必要的最新版本包
pip install --upgrade setuptools wheel twine
创建新版本的源和二进制发行版
python setup.py sdist bdist_wheel && twine check dist/*
修复您遇到的任何错误。
将源发行版上传到PyPI
twine upload dist/*
提交任何挂起的更改
git commit -a git push
在GitHub上将项目的最新版本标记为来自setup.py文件的版本号。例如,如果setup.py中的版本号为0.0.1,则执行
git tag 0.0.1 git push --tags
项目详情
ckanext-xloader-0.12.2.tar.gz的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | c7f347b6bd038c7b054f7e91a54159a6438e506cb7a284852f2566ade98b8f4b |
|
MD5 | 65c5fd5bd3c5a403257fb0c50f9bf6f6 |
|
BLAKE2b-256 | a4a890d9d58a3d6e8411fa9543830ce21129de9904d57c114029b6521025caef |