datapackage-pipelines 2.1.9

类似的结果可以通过定义正则表达式而不是常量来实现

$ pip install datapackage-pipelines

转储到SQL

将数据包保存到SQL数据库。

参数：

• 引擎-用于连接到SQL数据库的连接字符串（URL语法） 还支持env://<；环境变量>；，它表示应该从指定的环境变量中获取连接字符串。 如果未指定，则假定默认值为env://dpp_db_engine
• 表-资源与数据库表之间的映射。键是表名，值是具有以下属性的对象：
  • 资源名称-应转储到表中的资源的名称
  • 模式如何将数据写入数据库。 可能值：
    • 重写（默认值）-重写表，将删除所有以前的数据（如果有）。
    • 追加-在不更改现有数据的情况下写入新行。
    • 更新-根据一组"更新键"更新表。 对于每个新行，查看数据库中是否已经存在可以更新的现有行（即现有行 所有更新键中的值都相同）。 如果是，则更新现有行中的其余列。否则-在数据库中插入新行。
  • 更新键-仅适用于更新模式。用于检查行是否存在的字段名列表。 如果未指定，则将使用架构的primarykey作为默认值。
  • 索引-tbd
• updated_column-用布尔值添加到spewed数据的列的可选名称
  • true-行已更新
  • false-已插入行
• updated_id_column-将添加到spewed数据并包含数据库中已更新行的id的列的可选名称。

转储到路径

将数据包保存到文件系统路径。

参数：

• 输出路径-存储datapackage.json的输出路径的名称。

  如果该路径不存在，则将创建该路径以及内部数据包路径。

  如果省略，则假定为（当前目录）。

• 强制格式-指定是否强制以相同格式生成所有输出文件

  • 如果true（默认值），则所有资源都将使用相同的格式
  • 如果false，则将从文件扩展名推导格式。具有未知扩展名的资源将被丢弃。

• format-指定要生成的输出文件类型（如果force format为true）：csv（默认值）或json

• 将文件哈希添加到路径：指定是否将文件md5哈希包含到资源路径中。默认为false。如果true将散列嵌入如下路径：

  • 如果原始路径是路径/to/the/file.ext
  • 修改后的路径将是path/to/the/hash/file.ext

• 计数器-指定是对数据的行、字节还是MD5哈希进行计数，以及数据应存储在哪里。具有以下属性的对象：

  • datapackage row count：数据包的总行计数应存储在何处（默认值：count/u rows）
  • 数据包字节：数据包的总字节计数应存储在哪里（默认值：字节）
  • 数据包哈希：数据包的md5哈希应存储在哪里（默认值：哈希）
  • 资源行数：每个资源的总行数应存储在哪里（默认值：行数）
  • 资源字节：每个资源的总字节计数应存储在哪里（默认值：字节）
  • 资源哈希：每个资源的md5哈希应存储在哪里（默认值：哈希） 每个属性都可以ld设置为空以防止计数。 每个属性都可以是点分隔的字符串，用于将数据存储在嵌套对象中（例如stats.rowcount）

• pretty descriptor：指定数据包描述符（datapackage.json）文件的外观：

  • false（默认）-描述符将写入一行。
  • true-描述符对每个键都有缩进和新行，因此它变得更易于阅读。

转储到邮政编码

将数据包保存到压缩存档。

参数：

• 输出文件-压缩数据将存储的输出文件的名称
• 强制格式和格式-与转储到路径中相同
• 将文件哈希添加到路径-与dump\u to\u path中的相同
• 计数器-与转储到路径中的相同
• 漂亮的描述符-与"转储到"路径中的相同

不推荐使用的处理器

这些处理器将在下一个主要版本中删除。

添加元数据

由于向后兼容的原因，保留了update_package的别名。

添加资源

将新的外部表格资源添加到数据包。

参数：

您应该提供a名称和URL属性，以及其他可选属性，这些属性是在规范中定义的。

url指示此资源的数据所在的位置。稍后，当stream_remote_resources运行时，它将使用url（存储在dpp:streamedfrom属性的资源中）读取数据行并将其推入管道。

请注意，url还支持env://<；环境变量>；，这表示应该从指定的环境变量获取资源url。这在为字符串提供敏感信息（例如用于从数据库表流式处理的SQL连接字符串）时非常有用。

参数基本上是传递给atabulator.stream实例的参数（请参见api引用rel="nofollow">api）。 除此之外，还可以传递一个常量参数，该参数应该是头到字符串值的映射。 当与stream_remote_resources一起使用时，这些常量值将添加到每个生成的行中 （以及默认模式）。

您也可以在这里提供一个模式，或者使用由stream_remote_resources处理器生成的默认模式。 如果指定了路径，则将使用该路径。否则，stream_remote_resources处理器将为您分配路径和csv扩展名。

示例：

$ pip install datapackage-pipelines

6

流远程资源

将外部资源转换为流式资源。

外部资源是指链接到远程数据源（URL或文件路径）的资源，但不由管道处理并保持原样。

流式资源是可以由管道处理的资源，其输出作为结果数据包的一部分保存。

如果某个资源没有模式，则通过从数据源中的每一列创建字符串字段，在此处自动生成默认模式。

参数：

• 资源-要传输哪些资源。可以是：

  • 字符串列表，解释为资源名to流
  • 字符串，解释为用于匹配资源名称的正则表达式

  如果省略，则数据包中的所有资源都将流式传输。

• 忽略丢失-如果为true，则丢失的资源不会引发错误，而是将被视为"空"（即零行）。 具有空URL的资源将被视为相同的资源（即将生成"空"资源）。

• 限制行-如果提供，将限制从源中获取的行数。获取一个整数值，该整数值指定要传输的源的行数。

示例：

$ pip install datapackage-pipelines

7

此处理器还支持加载纯文本资源（例如HTML页面）并将其作为表格数据处理-使用单个"数据"列拆分成行。 要启用此行为，请将以下属性添加到资源："format"："txt"

转储到SQL

出于向后兼容的原因，保留了"转储到SQL"的别名。

转储到路径

将数据包保存到文件系统路径。

参数：

• 输出路径-存储datapackage.json的输出路径的名称。

  如果该路径不存在，则将创建该路径以及内部数据包路径。

  如果省略，则假定为（当前目录）。

• 强制格式-指定是否强制以相同格式生成所有输出文件

  • 如果true（默认值），则所有资源都将使用相同的格式
  • 如果false，则将从文件扩展名推导格式。具有未知扩展名的资源将被丢弃。

• format-指定要生成的输出文件类型（如果force format为true）：csv（默认值）或json

• handle non-tabular-指定是否应将非表格式资源（即没有架构的资源）转储到结果数据包。 （有关详细信息，请参见下面的注释）

• 将文件哈希添加到路径：指定是否将文件md5哈希包含到资源路径中。默认为false。如果true将散列嵌入如下路径：

  • 如果原始路径是路径/to/the/file.ext
  • 修改后的路径将是path/to/the/hash/file.ext

• 计数器-指定是对数据的行、字节还是MD5哈希进行计数，以及数据应存储在哪里。具有以下属性的对象：

  • datapackage row count：数据包的总行计数应存储在何处（默认值：count/u rows）
  • 数据包字节：数据包的总字节计数应存储在哪里（默认值：字节）
  • 数据包哈希：数据包的md5哈希应存储在哪里（默认值：哈希）
  • 资源行数：每个资源的总行数应存储在哪里（默认值：行数）
  • 资源字节：每个资源的总字节计数应存储在哪里（默认值：字节）
  • 资源哈希：每个资源的md5哈希应存储在哪里（默认值：哈希） 为了防止计数，这些属性中的每一个都可以设置为空。 每个属性都可以是点分隔的字符串，用于将数据存储在嵌套对象中（例如stats.rowcount）

• pretty descriptor：指定数据包描述符（datapackage.json）文件的外观：

  • false（默认）-描述符将写入一行。
  • true-描述符对每个键都有缩进和新行，因此它变得更易于阅读。

• 文件格式化程序：指定自定义文件格式处理程序。一个具有格式名到python模块和类名映射的对象。

  • 允许重写现有的csv和json格式处理程序或添加对新格式的支持。
  • 请注意，这些更改可能会使生成的数据包与FrictionLessData规范不兼容，并可能导致互操作性问题。
  • 示例用法：pipeline-spec.yaml（在自定义格式化程序管道下），xlsxformat类

转储.to_zip

将数据包保存到压缩存档。

参数：

• 输出文件-压缩数据将存储的输出文件的名称
• 强制格式和格式-与转储到路径中相同
• 处理非表格格式-与转储到路径中的相同
• 将文件哈希添加到路径-与dump\u to\u path中的相同
• 计数器-与转储到路径中的相同
• 漂亮的描述符-与"转储到"路径中的相同
• 文件格式化程序-与"转储到"路径中的相同

注意

dump.to_path和dump.to_zip处理器也将处理非表格资源。 这些资源必须同时具有url和path属性，并且不能包含schema属性。 在这种情况下，将从url下载文件，并将其放置在提供的路径中

自定义处理器

对于任何非琐碎的处理任务，您可能会遇到使用标准库处理器无法解决的问题，这是非常合理的。

为此，您可能需要编写自己的处理器-以下是处理方法。

有两种用于编写处理器的API—高级API和低级API。

高级处理器API

高级api对大多数处理器类型都非常有用：

$ pip install datapackage-pipelines

8

高级api由一个方法process组成，它有两个函数：

• 修改数据包-对数据包描述符进行更改（如有必要），例如添加元数据、添加资源、修改资源的架构等。

  必要时也可用于初始化代码。

  它有以下参数：

  • data package是需要修改的当前数据包描述符。 需要返回修改后的数据包描述符。
  • parameters是包含处理器参数的dict，如pipeline-spec.yaml文件中提供的。
  • stats是一个dict，为了收集过程中的度量和测量（例如验证检查、行计数等），应该对其进行修改。

• 处理行-修改流中的一行。它接收这些参数：

  • 行是包含要处理的行的字典
  • 行索引是资源中该行的索引
  • 资源描述符是当前正在处理的资源的描述符对象
  • 资源索引是数据包中资源的索引
  • parameters是包含处理器参数的dict，如pipeline-spec.yaml文件中提供的。
  • stats是一个dict，为了在过程（例如验证检查、行计数等）

  并生成零个或多个已处理行。

几个例子

$ pip install datapackage-pipelines

9

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

0

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

1

低级处理器API

在某些情况下，高级api可能过于受限。在这种情况下，您应该考虑使用低级api。

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

2

上面的代码片段显示了大多数低级处理器的结构。

我们总是从调用insert（）开始-此方法为我们提供上下文，其中包含执行参数、数据包描述符（从上一步输出）和所有流式资源行上的迭代器。

我们通过调用spew（）来完成处理，后者将处理后的数据发送到管道中的下一个处理器。spew接收：

• 修改的数据包描述符；
• 资源上的（可能是新的）迭代器；
• 一个stats对象，它将被添加到前面步骤中的stats中，并在管道完成时返回给用户，并且；
• （可选）一个finalizer函数，它在完成对资源的迭代之后，但在向其他处理器发送完成的信号之前将被调用。例如，您可以使用它关闭任何打开的文件。

更深入的解释

spew按以下顺序写入接收到的数据：

• 首先，将datapackage参数写入流。 这意味着对数据包描述符的所有修改都必须在调用spew之前完成。 一个常见的陷阱是修改资源迭代器中的数据包描述符-尽量避免这样做，因为下一个处理器将接收到的描述符将是错误的。
• 然后它开始迭代资源。对于每个资源，它在其行上迭代并将每一行写入流。 这个迭代过程最终导致对原始资源迭代器（从摄取返回的资源迭代器）的迭代。反过来，这会导致读取进程的输入流。由于操作系统中缓冲的工作方式，"慢"处理器将缓慢读取其输入，导致之前的处理器在其CPU密集型处理器完成处理时在IO上休眠。"快速"处理器不会漫无目的地工作，而是在等待输入数据或等待输出缓冲区耗尽时休眠。 这里所得到的结果是，数据中的所有行都或多或少地同时被处理，并且没有一个处理器在可能在后续处理步骤中失败的行上工作得太"超前"。
• 然后将统计数据写入流。这意味着可以在迭代期间修改统计信息，并且只使用迭代完成后的值。
• 最后，调用finalizer方法（如果我们收到一个）。

几个例子

我们将从上面的相同处理器开始，现在用低级API实现。

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

3

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

4

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

5

下一个例子展示了如何实现一个简单的web scraper。虽然不是严格要求，但web刮板通常是管道中的第一个处理器。因此，它们可以忽略传入的数据包和资源迭代器，因为没有以前的处理器生成数据：

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

6

在这个例子中，我们可以看到初始的数据包是从头开始生成的，而资源迭代器实际上是一个scraper，在从ckan实例api接收行时生成它们。

插件和源描述符

在特定问题域中编写管道时，可能会发现开发的处理管道遵循某种模式。抓取或获取源数据往往是相似的。处理、数据清理和验证通常是相同的。

为了便于维护和避免样板，可以编写数据包管道插件。

插件是名为datapackage\u pipelines\lt；plugin name>；的python模块。插件可以提供两种功能：

• 处理器包-您可以在插件中打包围绕某个主题或特定用途的处理器。任何位于数据包管道plugin name>；下的处理器foo都可以在管道内用作<；plugin name>；.foo模块。
• 管道模板-如果类generator存在于datapackage\u pipelines\lt；plugin name>；模块中，则它将用于基于模板生成管道-我们称之为"源描述符"。

源描述符

源描述符是包含用于创建完整管道的信息的yaml文件。

dpp将查找名为<；plugin name>；.source-spec.yaml的文件，并将它们视为管道生成代码的输入-应在datapackage\u pipelines\lt；plugin name>；中名为generator的类中实现；模块。

这个类应该继承自generatorbase并且应该实现两个方法：

• 生成管道- 它接收源描述并返回形式为（id，details）的元组迭代器。 id可能是管道id，在这种情况下，详细信息将是包含管道定义的对象。 如果id的格式为：module:，则详细信息将被视为来自指定模块的源规范。这样，生成器可能会生成其他源规范。
• 获取模式-它应该返回一个json模式，用于验证源描述的结构

示例

假设我们编写了一个数据包管道插件，用于从ckan实例中提取数据。

下面是这样一个假设生成器的外观：

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

7

在这种情况下，如果我们存储一个类似这样的文件：

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

8

然后当运行dpp时，我们将看到一个名为/example com的可用管道包列表

这个管道在内部由3个步骤组成：ckan.scraper，元数据和转储到压缩文件

验证源描述符

源描述符可以具有与输出管道的参数域最匹配的任何结构。但是，它必须有一个一致的结构，由json模式文件支持。在我们的例子中，模式可能如下：

$ dpp
Available Pipelines:
- ./worldbank-co2-emissions (*)

$ $ dpp run --verbose ./worldbank-co2-emissions
RUNNING ./worldbank-co2-emissions
Collecting dependencies
Running async task
Waiting for completion
Async task starting
Searching for existing caches
Building process chain:
- update_package
- load
- set_types
- dump_to_zip
- (sink)
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/update_package.py
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :Starting new HTTP connection (1): api.worldbank.org:80
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
load: DEBUG   :http://api.worldbank.org:80 "GET /v2/en/indicator/EN.ATM.CO2E.PC?downloadformat=excel HTTP/1.1"200308736
set_types: INFO    :(<dataflows.processors.set_type.set_type object at 0x10a5c79b0>,)
load: INFO    :Processed 264 rows
set_types: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/load.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/set_types.py
dump_to_zip: INFO    :Processed 264 rows
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/manager/../lib/internal/sink.py
DONE /Users/adam/code/dhq/specstore/dpp_repo/datapackage_pipelines/specs/../lib/dump_to_zip.py
DONE V ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}
INFO    :RESULTS:
INFO    :SUCCESS: ./worldbank-co2-emissions {'bytes': 692741, 'count_of_rows': 264, 'dataset_name': 'co2-emissions', 'hash': '4dd18effcdfbf5fc267221b4ffc28fa4'}

9

dpp将确保源描述符文件在尝试使用generator类将其转换为管道之前符合该架构。

提供处理器代码

在某些情况下，生成器也希望提供处理器代码（与管道定义一起）。 为此，生成器可以向包含处理器代码的任何步骤添加code属性。执行此步骤时，不会像往常一样尝试解析处理器，而是会改为解析提供的代码。

按计划运行

datapackage pipelines带有芹菜式集成，允许通过类似于crontab的语法在特定时间运行管道。

为了实现这一点，您只需将schedule部分添加到pipeline-spec.yaml文件中（或者从生成器类返回一个schedule，见上文），如下所示：

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

0

在本例中，此管道设置为每小时运行一次，每小时运行一次。

要运行芹菜守护进程，请使用celeri的命令行界面运行datapackage\u pipelines.app。有一种方法

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

1

运行此服务器将从执行所有"脏"任务开始，并根据任务的计划继续执行任务。

作为启动调度程序和仪表板的快捷方式（见下文），您可以使用预先构建的Docker图像：

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

2

然后浏览到http://<；docker machine的ip address>；：5000/查看当前执行状态仪表板。

管道仪表板和状态标牌

当安装在服务器上或使用任务调度程序运行时，通常很难确切地知道每个管道的运行情况和状态。

为了简化操作，您可以旋转Web仪表板，以提供每个管道的状态、基本信息和最新执行结果的概述。

要启动web服务器，请从命令行运行dpp serve并浏览到http://localhost:5000

环境变量dpp_base_path将确定仪表板是从根路径还是从另一个基本路径（示例值：/pipelines/）提供服务。

仪表板端点可以通过添加带有环境变量dpp_basic_auth_username和dpp_basic_auth_password的用户名和密码来要求验证。

对于单个管道和管道集合，甚至更简单的管道状态都可以使用状态标记。对于单个管道，将完整的管道ID添加到徽标端点：

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

3

或者对于管道集合：

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

4

请注意，无论dpp_basic_auth_password和dpp_basic_auth_username设置如何，这些徽章端点都将始终公开。

与其他服务集成

数据包管道可以对任何管道事件调用预定义的webhook。这可能允许潜在的集成与其他应用程序一起使用。

要在特定管道中添加Webhook，请在管道定义中添加hooks属性，该属性应该是URL列表。 每当该管道排队、开始运行或结束运行时，所有URL都将使用此负载进行发布：

$ docker run -it -v `pwd`:/pipelines:rw \
        frictionlessdata/datapackage-pipelines
<available-pipelines>

$ docker run -it -v `pwd`:/pipelines:rw \
       frictionlessdata/datapackage-pipelines run ./worldbank-co2-emissions
<execution-logs>

5

已知问题

• 加载在单个单元格中包含大量数据的资源时会引发异常（\112）

标签：

• org
• load
• description
• 资源
• co2
• pipelines
• datapackage
• dpp
• emissions

欢迎加入QQ群-->： 979659372

推荐PyPI第三方库

• dist-gb

  高斯分布

• responseSpect

  加速度、速度和位移反应谱计算

• festin

  Festin是一个强大的S3桶查找和内容发现

• labeltext

  labeltext是一个简单的命令行实用程序，可以为文本分类任务快速注释大量文本。

• merossiot

  防止攻击的包

• resp3

  RESP3的python实现。

• ezephys

  处理电生理数据的工具。

• JS-Prob

  高斯分布

• anomaly-detection-framework

  异常检测框架允许我们计算任何时间序列数据集的异常。它有一个易于管理的接口，可以用给定的数据集训练预测。

• veracitysdk

  准确性api库。

• sparse-pendulum

  开放AI健身房摆锤环境与稀疏奖励

• additions

  对Python标准库的添加

• vcap

  用于在Python中创建OpenVisionCapsules的库

• jsconfp

  json save config是一个简单的配置文件库

• assert-json-schema

  断言json架构，还提供$参考：相关的_文件包装到$参考：文件：{绝对路径}