问题标签 [data-pipeline]

我一直在尝试多种方法将压缩的 TSV 移动到大查询。我能够使命令正常工作，但没有看到正在加载任何表。请帮我弄清楚编写有效的命令。

bq '--project_id' --nosync load --source_format CSV --field_delimiter '\t' --autodetect --skip_leading_rows '0' --quote='' --encoding UTF-8 :table.destinationtable 'gs:/ /bucketname/filename.tsv.gz'</p>

成功开始加载162822：bqjob_r2d00a5817904935f_0000015c79e61b7c_1

我在 Azure Blob 存储中有一堆文件，并且不断有新文件。我想知道是否有办法让我首先获取 Blob 中的所有数据并将其移至 BigQuery，然后保持脚本或某些作业运行，以便将其中的所有新数据发送到 BigQuery？

我们正在使用 Kafka、storm 和 redis 开发数据管道应用程序。来自不同系统的实时事件将发布到 Kafka，storm 根据配置的规则进行事件处理。状态在 redis 中管理。

我们需要在处理不同的事件之前实现不同的 WAIT_TIME。我们正在研究以下选项。

我们最初查看了风暴窗口 [滑动或翻滚窗口]，但仅提供配置固定间隔的选项。我们需要根据规则改变 wait_time

我们正在探索将事件存储在 redis 缓存中不同持续时间 [TTL] 的其他选项，一旦每个事件被驱逐，我们需要回调到 Storm 来处理它。

redis 是否支持驱逐回调？有没有更好的方法来使用storm和redis？

我正在尝试在云形成堆栈中运行数据管道。此堆栈引用另一个包含 Redshift 集群的堆栈的导出。当我运行它时，我收到一条错误消息，指出“'Ec2Instance'，错误 = 验证此对象期间的内部错误。”；但我无法找到有关此错误或其含义的更多信息。其他对象显示相同的错误, 但它们依赖于 Ec2Instance。基于此，我假设这是级联情况。这是我的 Ec2Instance 的 PipelineObject。

我对数据管道和无服务器都是新手。我想知道如何使用无服务器自动化 AWS 数据管道。下面是我将 dynamo db 表导出到 S3 的 AWS 数据管道图

张量板图

我正在运行此代码以对来自 Google Streetview 的门牌号进行分类，它运行然后在第一步中进入某种循环，我不知道为什么。

我已将问题缩小到输入管道。

我已将链接附加到我存储数据和代码的 Google 云端硬盘：

https://drive.google.com/open?id=0B-hAFmA-zmGdTndyaHJzWEdQaFE

它包含三个文件。

• model.ipynb - 它是包含 CNN 和会话的主文件，
• preprocess.ipynb - 这是我用来处理我的数据和
• data.pickle - 我已经在这个文件中处理并存储了我的数据，所以你不必运行 preprocess.ipynb。

我是 AWS 新手。我通过 AWS 控制台使用 Auto Scaling 策略创建了一个 EMR 集群。我还创建了一个数据管道，可以使用这个集群来执行活动。

我还可以通过数据管道动态创建 EMR 集群。但是在执行此操作时，我无法将 Auto Scaling Rule 分配给 EMR 集群。有没有办法通过数据管道为 EMR 集群配置 Auto Scaling 角色和其他所需的配置

情况：我有 2 个按需运行的数据管道。在管道 A 完成之前，管道 B 无法运行。我正在尝试在单个脚本/程序中自动运行两个管道，但我不确定如何在 Go 中完成所有这些工作。

我有一些激活数据管道的 Go 代码：

激活后，我希望能够监视该管道并确定它何时完成，以便我可以运行第二个管道。类似于list-runsCLI 命令，但我不确定相应的 Go 函数是什么。

因此，一旦所有操作都标记为“已完成”，我想激活我的第二个管道。实现这一目标的最佳方法是什么？

我正在学习气流并且有一个简单的问题。下面是我的 DAG，名为dog_retriever：

作为测试 Airflow 的一种方法，我只是在这个非常简单的http://dog.ceo API 中向某些端点发出了两个 GET 请求。目标是学习如何处理通过 Airflow 检索到的一些数据

执行正在运行——我的代码成功调用了任务 t1 和 t2 中的端点，我可以看到它们以基于set_upstream我编写的规则的正确顺序记录在 Airflow UI 中。

我不知道如何访问这两个任务的 JSON 响应。这似乎很简单，但我无法弄清楚。在 SimpleHtttpOperator 中，我看到了response_check的参数，但没有任何东西可以简单地打印、存储或查看 JSON 响应。

谢谢。

我在 python 中遇到了 psycopg2 的一些问题
我有两个与相应游标不同的连接：

假设有一个我想在 source_cursor 上执行的选择查询：

此查询的结果应插入到 dest_cursor 的表中

我研究了copy_from和copy_to以及诸如此类的示例。他们使用外部文件来复制数据，而我需要直接复制而不使用文件。源数据库和目标数据库之间的直接连接是不可能的，否则我会在数据库端完成。

我认为可以使用fetchall()和INSERT.. 或直接使我们能够复制数据的某种方法。

非常感谢任何有关类似示例链接的帮助