标签: amazon-emr

我想运行像这样的陈述

SELECT date_add('2008-12-31', 1) FROM DUAL

Hive(在Amazon EMR上运行)有类似的东西吗？

有没有人知道用于Amazon Web Services的Scala SDK？我对EMR工作特别感兴趣.

我需要在一个DynamoDB表中执行大约1.3亿个项目(总共5个Gb)的初始上传.在我使用应用程序中的API上传问题后,我决定尝试使用EMR.

长话短说,即使在最强大的集群上,导入非常平均(对于EMR)的数据量也需要很长时间,耗费数百小时且进展很少(大约20分钟处理测试2Mb数据位,并且没有管理在12小时内完成测试700Mb文件).

我已经联系了亚马逊高级支持,但到目前为止他们只是说"出于某种原因,DynamoDB导入很慢".

我在交互式蜂巢会话中尝试了以下说明:

CREATE EXTERNAL TABLE test_medium (
  hash_key string,
  range_key bigint,
  field_1 string,
  field_2 string,
  field_3 string,
  field_4 bigint,
  field_5 bigint,
  field_6 string,
  field_7 bigint
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
LOCATION 's3://my-bucket/s3_import/'
;

CREATE EXTERNAL TABLE ddb_target (
  hash_key string,
  range_key bigint,
  field_1 bigint,
  field_2 bigint,
  field_3 bigint,
  field_4 bigint,
  field_5 bigint,
  field_6 string,
  field_7 bigint
)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES (
  "dynamodb.table.name" = "my_ddb_table",
  "dynamodb.column.mapping" = "hash_key:hash_key,range_key:range_key,field_1:field_1,field_2:field_2,field_3:field_3,field_4:field_4,field_5:field_5,field_6:field_6,field_7:field_7"
)
;  

INSERT OVERWRITE TABLE …

我正在尝试使用EMR/Hive将数据从S3导入DynamoDB.我的CSV文件包含用双引号括起来并用逗号分隔的字段.在hive中创建外部表时,我可以将分隔符指定为逗号但是如何指定字段用引号括起来？

如果我没有指定,我看到DynamoDB中的值填充在两个双引号""value""中,这似乎是错误的.

我正在使用以下命令来创建外部表.有没有办法指定字段用双引号括起来？

CREATE EXTERNAL TABLE emrS3_import_1(col1 string, col2 string, col3 string, col4 string)  ROW FORMAT DELIMITED FIELDS TERMINATED BY '","' LOCATION 's3://emrTest/folder';

任何建议,将不胜感激.谢谢Jitendra

我在写一个镶木地板文件时遇到这个错误,这已经开始发生了

com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.services.s3.model.AmazonS3Exception: Please reduce your request rate. (Service: Amazon S3; Status Code: 503; Error Code: SlowDown; Request ID: 2CA496E2AB87DC16), S3 Extended Request ID: 1dBrcqVGJU9VgoT79NAVGyN0fsbj9+6bipC7op97ZmP+zSFIuH72lN03ZtYabNIA2KaSj18a8ho=
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:1389)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java:902)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:607)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.doExecute(AmazonHttpClient.java:376)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.executeWithTimer(AmazonHttpClient.java:338)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:287)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3826)
    at com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.services.s3.AmazonS3Client.deleteObjects(AmazonS3Client.java:1777)
    at com.amazon.ws.emr.hadoop.fs.s3.lite.call.DeleteObjectsCall.perform(DeleteObjectsCall.java:22)
    at com.amazon.ws.emr.hadoop.fs.s3.lite.call.DeleteObjectsCall.perform(DeleteObjectsCall.java:7)
    at com.amazon.ws.emr.hadoop.fs.s3.lite.executor.GlobalS3Executor.execute(GlobalS3Executor.java:75)
    at com.amazon.ws.emr.hadoop.fs.s3.lite.AmazonS3LiteClient.invoke(AmazonS3LiteClient.java:176)
    at com.amazon.ws.emr.hadoop.fs.s3.lite.AmazonS3LiteClient.deleteObjects(AmazonS3LiteClient.java:125)
    at com.amazon.ws.emr.hadoop.fs.s3n.Jets3tNativeFileSystemStore.deleteAll(Jets3tNativeFileSystemStore.java:355)
    at sun.reflect.GeneratedMethodAccessor121.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:191)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
    at com.sun.proxy.$Proxy28.deleteAll(Unknown Source)
    at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.doSingleThreadedBatchDelete(S3NativeFileSystem.java:1331)
    at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.delete(S3NativeFileSystem.java:663)
    at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.delete(EmrFileSystem.java:296)
    at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.cleanupJob(FileOutputCommitter.java:463)
    at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.abortJob(FileOutputCommitter.java:482)
    at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.abortJob(HadoopMapReduceCommitProtocol.scala:134)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala:146) …

我有一个火花工作,我在两个数据帧之间进行外连接.第一个数据帧的大小为260 GB,文件格式为文本文件,分为2200个文件,第二个数据帧的大小为2GB.然后将大约260 GB的数据帧输出写入S3需要很长时间超过2小时后我取消因为我已经在EMR上进行了大量更改.

这是我的群集信息.

emr-5.9.0
Master:    m3.2xlarge
Core:      r4.16xlarge   10 machines (each machine has 64 vCore, 488 GiB memory,EBS Storage:100 GiB)

这是我正在设置的群集配置

capacity-scheduler  yarn.scheduler.capacity.resource-calculator :org.apache.hadoop.yarn.util.resource.DominantResourceCalculator
emrfs-site  fs.s3.maxConnections:   200
spark   maximizeResourceAllocation: true
spark-defaults  spark.dynamicAllocation.enabled:    true

我尝试手动设置内存组件,如下所示,性能更好,但同样的事情又花了很长时间

--num-executors 60 - conf spark.yarn.executor.memoryOverhead = 9216 --executor-memory 72G --conf spark.yarn.driver.memoryOverhead = 3072 --driver-memory 26G --execeror-cores 10 - driver-cores 3 --conf spark.default.parallelism = 1200

我没有使用默认分区将数据保存到S3.

添加有关作业和查询计划的所有详细信息,以便于理解.

真正的原因是分区.这大部分时间都在占用.因为我有2K文件,所以如果我使用像200这样的重新分区,输出文件以十万分之一形式出现,然后在spark中重新加载并不是一个好故事.

在下图中我不知道为什么在项目之后再次调用sort

在下面Image GC对我来说太高了..请oi必须处理这个请建议如何？

下面是节点健康状态.这一点数据被保存到S3中,难怪为什么我只能看到两个节点处于活动状态并且都处于空闲状态.

这是加载时的集群细节.在这一点上,我可以看到集群已被充分利用,但在将数据保存到S3时,许多节点都是免费的.

最后这里是我的代码,我执行Join然后保存到S3 ...

import org.apache.spark.sql.expressions._

          val windowSpec = Window.partitionBy("uniqueFundamentalSet", "PeriodId", "SourceId", "StatementTypeCode", "StatementCurrencyId", "FinancialStatementLineItem_lineItemId").orderBy(unix_timestamp($"TimeStamp", …

这与这个问题非常接近，但是我添加了一些特定于我的问题的细节：

使用AWS-EMR jupyter笔记本进行Matplotlib绘图

我想找到一种在Jupyter笔记本中使用matplotlib的方法。这是错误的代码片段，非常简单：

笔记本

import matplotlib
matplotlib.use("agg")
import matplotlib.pyplot as plt
plt.plot([1,2,3,4])
plt.show()

我选择此代码段是因为仅此行在尝试使用TKinter时失败（该行未安装在AWS EMR集群中）：

import matplotlib.pyplot as plt

当我运行完整的笔记本代码片段时，结果是没有运行时错误，但什么也没有发生（未显示任何图形。）我对这可以起作用的一种方式的理解是添加以下两个代码片段之一：

pyspark魔术符号

%matplotlib inline

结果

unknown magic command 'matplotlib'
UnknownMagic: unknown magic command 'matplotlib'

IPython显式魔术调用

unknown magic command 'matplotlib'
UnknownMagic: unknown magic command 'matplotlib'

结果

'NoneType' object has no attribute 'run_line_magic'
Traceback (most recent call last):
AttributeError: 'NoneType' object has no attribute 'run_line_magic'

到我的笔记本，笔记本会调用一个火花魔术命令，该命令会内联matplotlib图（至少是我的解释。）我在使用引导操作后尝试了这两种方法：

EMR引导程序

from IPython import get_ipython
get_ipython().run_line_magic('matplotlib', 'inline')

即使添加了这些，我仍然会得到一个错误，那就是matplotlib没有任何魔力。所以我的问题肯定是：

题

如何使matplotlib在AWS EMR Jupyter笔记本中工作？

（或者如何在AWS EMR Jupyter笔记本中查看图形并绘制图像？）

我正在尝试启动集群并使用boto运行所有作业.我找到了许多创建job_flows的例子.但我不能为我的生活找到一个例子,表明:

1. 如何定义要使用的集群(通过clusted_id)
2. 如何配置启动集群(例如,如果我想为某些任务节点使用点实例)

我错过了什么吗？

我正在使用亚马逊的具体星火的EMR集群(版本EMR-4.2.0)maximizeResourceAllocation标志作为记录在这里.根据这些文档,"此选项计算核心节点组中节点上执行程序可用的最大计算和内存资源,并使用此信息设置相应的spark-defaults设置".

我正在使用m3.2xlarge实例为工作节点运行集群.我正在为YARN master使用一个m3.xlarge - 我可以运行它的最小m3实例,因为它没有做太多.

情况是这样的:当我运行Spark作业时,每个执行程序所请求的核心数是8.(我在配置之后才得到这个,"yarn.scheduler.capacity.resource-calculator": "org.apache.hadoop.yarn.util.resource.DominantResourceCalculator"这实际上不在文档中,但我离题了).这似乎是有道理的,因为根据这些文档,m3.2xlarge有8个"vCPU".但是,在实际实例本身中/etc/hadoop/conf/yarn-site.xml,每个节点都配置为yarn.nodemanager.resource.cpu-vcores设置为16.我(猜测)认为这必定是因为超线程或者其他一些硬件的原因.

所以问题在于:当我使用时maximizeResourceAllocation,我获得了亚马逊实例类型具有的"vCPU"数量,这似乎只是YARN在节点上运行的已配置"VCores"数量的一半; 因此,执行程序仅使用实例上实际计算资源的一半.

这是Amazon EMR中的错误吗？其他人是否遇到同样的问题？是否还有其他一些我缺少的魔法无证配置？

最近我们从"EMR on HDFS" - >"EMR on S3"(启用了一致视图的EMRFS)迁移,我们意识到与HDFS相比,写入S3的Spark'SaveAsTable'(镶木地板格式)速度慢约4倍但我们发现使用DirectParquetOutputCommitter - [1] w/Spark 1.6的解决方法.

S3缓慢的原因 - 我们不得不支付所谓的Parquet tax- [2],其中默认输出提交者写入临时表并稍后将其重命名为S3中的重命名操作非常昂贵

我们也了解使用'DirectParquetOutputCommitter'的风险,这可能会导致数据损坏,并启用推测任务.

现在w/Spark 2.0这个类已被弃用,我们想知道我们在桌面上有什么选项,这样我们就不会在升级到Spark 2.0时承受大约4倍的写入速度.任何想法/建议/建议都将受到高度赞赏.

我们可以想到的一个解决方法是 - 保存在HDFS上,然后通过s3DistCp将其复制到S3(当我们的Hive元数据存储指向S3时,如何以理智的方式完成这些任何想法？)

看起来像NetFlix已经修复了这个 - [3],有什么想法他们何时计划开源呢？

谢谢.

[1] - https://github.com/apache/spark/blob/21d5ca128bf3afd5c2d4c7fcc56240e28443474f/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/DirectParquetOutputCommitter.scala

[2] - https://www.appsflyer.com/blog/the-bleeding-edge-spark-parquet-and-s3/

[3] - https://www.youtube.com/watch?v=85sew9OFaYc&feature=youtu.be&t=8m39s http://www.slideshare.net/AmazonWebServices/bdt303-running-spark-and-presto-on-the -netflix -大数据平台