问题标签 [amazon-data-pipeline]

我希望在 Amazon Redshift 上的数据库中复制数据。在此之前，我将数据从 Redshift 数据库复制到托管在 EC2 实例上的 PostgreSQL 以进行分析。我有 ruby​​ 脚本，可以使用 dblink EXTENSION 来完成。

但现在由于数据在 PostgreSQL 实例上无法管理，我们决定将数据复制到同一 Redshift 集群上的单独数据库中。

撞！我发现 dblink 不适用于 Postgre 8.0（Redshift 版本）。有没有办法让我在 Redshift 上工作。

我不想使用 Copy 命令并在中间甚至数据管道之间添加一层 S3。你看，它是同一个集群，只是不同的数据库。

万一没有其他办法，我应该使用数据管道还是复制到 S3，然后再回到 Redshift。还是最终本质上是一样的？

PS任何帮助/线索将不胜感激。我会努力工作。只需要提个醒。

我想在 AWS DataPipeline 的控制台中定义一些参数，但无法这样做。这些参数将在 SqlActivity 中调用，因此当我尝试在内联 SQL 脚本中引用它们并保存管道时，我收到错误消息，指出不存在此类参数。参数页面是空白的，我找不到制作一个的按钮。谷歌结果都向我展示了如何创建参数化管道，这不是我想要的

我必须将我的 DynamoDb 表备份到 S3 中，但是当我启动此服务时，我在尝试三次后收到此错误：

private.com.amazonaws.AmazonServiceException：用户：arn:aws:sts::769870455028:assumed-role/DataPipelineDefaultResourceRole/i-3678d99c 无权执行：elasticmapreduce:ModifyInstanceGroups（服务：AmazonElasticMapReduce；状态代码：400；错误代码： AccessDeniedException；请求 ID：9065ea77-0f95-11e5-8f35-39a70915a1ef) 在 private.com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:1077) 在 private.com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java :725) 在 private.com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:295) 在 private.com.amazonaws.services 的 private.com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:460)。 elasticmapreduce.AmazonElasticMapReduceClient.invoke(AmazonElasticMapReduceClient.java:1391) 在 private.com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient.modifyInstanceGroups(AmazonElasticMapReduceClient.java:785) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606) 在 private.com.amazonaws.services.datapipeline.retrier.RetryProxy.invokeInternal(RetryProxy .java:36) 在private.com.amazonaws.services.datapipeline.retrier.RetryProxy.invoke(RetryProxy.java:48) 在 com.sun.proxy.$Proxy33.modifyInstanceGroups(Unknown Source) 在 amazonaws.datapipeline.cluster。 EmrUtil.acquireCoreNodes(EmrUtil.java:325) 在 amazonaws.datapipeline.activity.AbstractClusterActivity。resizeIfRequired(AbstractClusterActivity.java:47) at amazonaws.datapipeline.activity.AbstractHiveActivity.runActivity(AbstractHiveActivity.java:113) at amazonaws.datapipeline.objects.AbstractActivity.run(AbstractActivity.java:16) at amazonaws.datapipeline.taskrunner.TaskPoller .executeRemoteRunner(TaskPoller.java:132) at amazonaws.datapipeline.taskrunner.TaskPoller.executeTask(TaskPoller.java:101) at amazonaws.datapipeline.taskrunner.TaskPoller$1.run(TaskPoller.java:77) at private.com.amazonaws .services.datapipeline.poller.PollWorker.executeWork(PollWorker.java:76) 在 private.com.amazonaws.services.datapipeline.poller.PollWorker.run(PollWorker.java:53) 在 java.lang.Thread.run(Thread .java:745)runActivity(AbstractHiveActivity.java:113) at amazonaws.datapipeline.objects.AbstractActivity.run(AbstractActivity.java:16) at amazonaws.datapipeline.taskrunner.TaskPoller.executeRemoteRunner(TaskPoller.java:132) at amazonaws.datapipeline.taskrunner.TaskPoller .executeTask(TaskPoller.java:101) at amazonaws.datapipeline.taskrunner.TaskPoller$1.run(TaskPoller.java:77) at private.com.amazonaws.services.datapipeline.poller.PollWorker.executeWork(PollWorker.java:76)在 private.com.amazonaws.services.datapipeline.poller.PollWorker.run(PollWorker.java:53) 在 java.lang.Thread.run(Thread.java:745)runActivity(AbstractHiveActivity.java:113) at amazonaws.datapipeline.objects.AbstractActivity.run(AbstractActivity.java:16) at amazonaws.datapipeline.taskrunner.TaskPoller.executeRemoteRunner(TaskPoller.java:132) at amazonaws.datapipeline.taskrunner.TaskPoller .executeTask(TaskPoller.java:101) at amazonaws.datapipeline.taskrunner.TaskPoller$1.run(TaskPoller.java:77) at private.com.amazonaws.services.datapipeline.poller.PollWorker.executeWork(PollWorker.java:76)在 private.com.amazonaws.services.datapipeline.poller.PollWorker.run(PollWorker.java:53) 在 java.lang.Thread.run(Thread.java:745)TaskPoller.executeTask(TaskPoller.java:101) at amazonaws.datapipeline.taskrunner.TaskPoller$1.run(TaskPoller.java:77) at private.com.amazonaws.services.datapipeline.poller.PollWorker.executeWork(PollWorker.java:76 ) 在 private.com.amazonaws.services.datapipeline.poller.PollWorker.run(PollWorker.java:53) 在 java.lang.Thread.run(Thread.java:745)TaskPoller.executeTask(TaskPoller.java:101) at amazonaws.datapipeline.taskrunner.TaskPoller$1.run(TaskPoller.java:77) at private.com.amazonaws.services.datapipeline.poller.PollWorker.executeWork(PollWorker.java:76 ) 在 private.com.amazonaws.services.datapipeline.poller.PollWorker.run(PollWorker.java:53) 在 java.lang.Thread.run(Thread.java:745)

如何进行备份？有人有这个错误吗？谢谢

编辑：新政策

{“版本”：“2012-10-17”，“声明”：[{“效果”：“允许”，“操作”：[“s3：”，“dynamodb： ”，“ec2：Describe*”，“ elasticmapreduce:Describe*”、“elasticmapreduce:ListInstance*”、“elasticmapreduce:AddJobFlowSteps”、“elasticmapreduce: ”、“rds:Describe ”、 “datapipeline:” 、“cloudwatch:”、“redshift:DescribeClusters”、“redshift :DescribeClusterSecurityGroups"、"sdb: "、"sns: "、"sqs:" ], "资源": [ " " ] } ]

这是新的例外：

Error during job, obtaining debugging information... Examining task ID: task_1434014832347_0001_m_000008 (and more) from job job_1434014832347_0001 Examining task ID: task_1434014832347_0001_m_000013 (and more) from job job_1434014832347_0001 Examining task ID: task_1434014832347_0001_m_000005 (and more) from job job_1434014832347_0001 Examining task ID: task_1434014832347_0001_m_000034 (and more) from job job_1434014832347_0001 Examining task ID: task_1434014832347_0001_m_000044 (and more) from job job_1434014832347_0001 Examining task ID: task_1434014832347_0001_m_000004 (and more) from job job_1434014832347_0001 Task with the most failures(4): ----- Task ID: task_1434014832347_0001_m_000002网址：http://ip-10-37-138-149.eu-west-1.compute.internal:9026/taskdetails.jsp?jobid=job_1434014832347_0001&tipid=task_1434014832347_0001_m_000002 -----此任务的诊断消息：错误：Java堆space FAILED: 执行错误，从 org.apache.hadoop.hive.ql.exec.mr.MapRedTask MapReduce Jobs 返回代码 2

我注意到 AWS 数据管道的一些奇怪行为。是在Execution start time之前scheduled start time。请参阅下面的屏幕截图。

我在这里错过了什么吗？

这是 AWS 数据管道可接受的行为吗？避免这种情况的推荐方法是什么？

我知道 AWS Data Pipeline 支持并允许 SNS 警报，但我希望在查询返回任何内容时发送警报或电子邮件。基本上，我想运行一个非常简单的选择查询的 SQLActivity，如果该查询返回任何内容，我想发送电子邮件警报。这可能吗？如果是这样，我可以发送带有返回数据集的电子邮件吗？

我需要每隔几个小时备份 6 个 DynamoDB 表。我从模板创建了 6 个管道，它运行得很好，除了它创建了 6 个或更多的虚拟机，这些虚拟机大部分都在运行。这不是我能承受的经济。

有没有人有优化这种场景的经验？

我正在尝试将数据从 s3 移动到 redshift 中，并希望对 redshift 中的主键强制唯一性。我意识到复制命令本身不能做到这一点。但是，我注意到通过 AWS 数据管道可用的 RedshiftCopyActivity 允许使用“OVERWRITE_EXISTING”标志，这将允许至少以某种方式强制执行主键。

我想知道是否可以使用 boto 来实现这一点，以及是否有人可以为我指出这种用法的示例。

我刚刚在 Amazon AWS 上创建了一个帐户，我将使用 DATAPIPELINE 来安排我的查询。是否可以使用数据管道的 SQLACTIVITY 从 .sql 文件运行多个复杂的 SQL 查询？

我的总体目标是使用数据管道中的 sql 查询处理来自 REDSHIFT/s3 的原始数据并将其保存到 s3。这是可行的方法吗？

在这方面的任何帮助将不胜感激。

我想使用 AWS Data Pipeline 服务，并使用基于 JSON 的手动机制创建了一些服务，该机制使用 AWS CLI 创建、放置和激活管道。

我的问题是，如果管道定义发生变化，如何自动编辑或更新管道？我可以想象改变的事情可能是计划时间、活动或前提条件的添加或删除、对 DataNodes 的引用、资源定义等。

创建管道后，我们无法编辑官方文档中提到的很多内容：http: //docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-manage-pipeline-modify-console.html #dp-edit-pipeline-limits

这让我相信，如果我想自动更新管道，那么我必须删除并重新创建/激活一个新管道？如果是，那么下一个问题是如何创建一个自动化流程来识别先前版本的 ID，将其删除并创建一个新 ID？本质上是尝试为此构建一个发布管理流程，其中配置 JSON 文件会自动发布和部署。

大多数命令，如激活、删除、列表运行、put-pipeline-definition 等。使用管道 ID，直到创建新管道才知道。我找不到任何在更新或重新创建时保持不变的东西（createpipeline 命令的唯一 ID 和名称参数是一致的，但是我不能将它们用于上述任务（我需要管道 ID）。

当然，我可以尝试编写 grep 的 shell 脚本并搜索输出并尝试创建一个脚本，但还有其他更好的方法吗？我缺少的其他一些信息？

非常感谢。

谁能解释如何通过 AWS Lambda 执行使用源参数多次激活 AWS Data Pipeline。

目前，第一个处理激活管道，但随后任何后续激活都会失败，有或没有消息（取决于激活参数）

完整的处理循环是将数据移动到 S3，然后启动 DataPipeline 将数据迁移到 Redshift。

此摘录摘自 AWS 文档：

要激活已完成的管道，请修改管道的结束日期，然后将其激活。

AWS 文档参考