Dask EMR 无法读取 S3 存储桶中的数据库答案

【问题标题】：Dask EMR cant read in database in S3 bucketDask EMR 无法读取 S3 存储桶中的数据库
【发布时间】：2020-11-28 01:00:33
【问题描述】：

我使用以下代码创建了一个 Dask 集群并通过 ssh 连接到它

aws emr create-cluster --name "Dask-Cluster" \
--release-label emr-5.29.0 \
--applications Name=Hadoop \
--log-uri s3://aws-logs-045478813139-us-east-1/ \
--instance-type m5.xlarge \
--instance-count 3 \
--bootstrap-actions Path=s3://aws-emr-resources-045478813139-us-east-1/bootstrap-dask,Args="[--conda-packages,bokeh,fastparquet,python-snappy,snappy,matplotlib]" \
--use-default-roles \
--region us-east-1 \
--ec2-attributes '{"KeyName":"dask_key"}'

当我尝试读取位于 S3 中的 Amazon cmets parquet 数据库时，出现以下错误。我尝试加载示例数据库，但它也不起作用。但是，在本地运行或在此 EMR 集群中使用 Pandas 时，它可以工作。

我是云计算新手。任何帮助将不胜感激。

from dask_yarn import YarnCluster
from dask.distributed import Client

cluster = YarnCluster(environment="/home/hadoop/environment.tar.gz", worker_vcores=1, worker_memory="4GiB")

cluster.scale(8)
client = Client(cluster)

#read db    
df = dd.read_parquet('s3://amazon-reviews-pds/parquet/product_category=Books/*.parquet').compute()

---------------------------------------------------------------------------
KilledWorker                              Traceback (most recent call last)
<ipython-input-12-890ca19f35bc> in <module>()
----> 1 df = dd.read_parquet('s3://amazon-reviews-pds/parquet/product_category=Books/*.parquet').compute()

/home/hadoop/miniconda/lib/python3.7/site-packages/dask/base.py in compute(self, **kwargs)
    165         dask.base.compute
    166         """
--> 167         (result,) = compute(self, traverse=False, **kwargs)
    168         return result
    169 

/home/hadoop/miniconda/lib/python3.7/site-packages/dask/base.py in compute(*args, **kwargs)
    450         postcomputes.append(x.__dask_postcompute__())
    451 
--> 452     results = schedule(dsk, keys, **kwargs)
    453     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    454 

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2723                     should_rejoin = False
   2724             try:
-> 2725                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2726             finally:
   2727                 for f in futures.values():

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   1990                 direct=direct,
   1991                 local_worker=local_worker,
-> 1992                 asynchronous=asynchronous,
   1993             )
   1994 

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    831         else:
    832             return sync(
--> 833                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    834             )
    835 

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    338     if error[0]:
    339         typ, exc, tb = error[0]
--> 340         raise exc.with_traceback(tb)
    341     else:
    342         return result[0]

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/utils.py in f()
    322             if callback_timeout is not None:
    323                 future = asyncio.wait_for(future, callback_timeout)
--> 324             result[0] = yield future
    325         except Exception as exc:
    326             error[0] = sys.exc_info()

/home/hadoop/miniconda/lib/python3.7/site-packages/tornado/gen.py in run(self)
   1131 
   1132                     try:
-> 1133                         value = future.result()
   1134                     except Exception:
   1135                         self.had_exception = True

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1849                             exc = CancelledError(key)
   1850                         else:
-> 1851                             raise exception.with_traceback(traceback)
   1852                         raise exc
   1853                     if errors == "skip":

KilledWorker: ("('read-parquet-fa7aaf62acdade8f0210a8e768e4b232', 39)", <Worker 'tcp://172.31.27.94:35655', name: dask.worker_0, memory: 0, processing: 21>)

distributed.scheduler - INFO - Register worker <Worker 'tcp://172.31.27.94:43877', name: dask.worker_0, memory: 0, processing: 0>
distributed.scheduler - INFO - Starting worker compute stream, tcp://172.31.27.94:43877
distributed.core - INFO - Starting established connection

这些是运行读取命令之前来自客户端仪表板的工作日志：

我在集群详细信息中找不到任何日志文件

【问题讨论】：

请在您的 dask worker 和/或包含它的 EMR 容器中发布日志
谢谢@mdurant，我刚刚在帖子中添加了有关日志的信息。你说的是这些吗？对不起，如果我误解了。
旁白：我删除了 cloudflare-workers 标签，因为这个问题似乎与 Cloudflare Workers（来自不同公司的不同产品）无关。
我们确实需要来自工作人员的日志，而不是调度程序。我不知道那会在哪里。

标签： python amazon-s3 dask amazon-emr

【解决方案1】：

我建议的第一件事是确保集群中的工作人员具有访问 s3 存储桶的 IAM 权限 - 这可能意味着为集群分配一个角色。我会知道如何为 ec2 机器执行此操作，但不是 EMR 集群。

【讨论】：