标签: google-cloud-dataproc

我正在运行Spark作业（版本1.2.0），输入的内容是Google Clous存储桶中的文件夹（例如gs：// mybucket / folder）

在Mac机器上本地运行作业时，出现以下错误：

5932 [main]错误com.doit.customer.dataconverter.Phase1-日期作业：2014_09_23失败，错误：方案的无文件系统：gs

我知道要支持gs路径，需要做2件事。一种是安装GCS连接器，另一种是在Hadoop安装的core-site.xml中进行以下设置：

<property>
    <name>fs.gs.impl</name>
    <value>com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem</value>
    <description>The FileSystem for gs: (GCS) uris.</description>
</property>
<property>
    <name>fs.AbstractFileSystem.gs.impl</name>
    <value>com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS</value>
    <description>
     The AbstractFileSystem for gs: (GCS) uris. Only necessary for use with Hadoop 2.
    </description>
</property>

我认为我的问题来自这样一个事实，即我不确定在此本地模式下确切需要在哪里配置每个组件。在Intellij项目中，我正在使用Maven，因此我按以下方式导入了spark库：

<dependency> <!-- Spark dependency -->
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.10</artifactId>
    <version>1.2.0</version>
    <exclusions>
        <exclusion>  <!-- declare the exclusion here -->
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
        </exclusion>
    </exclusions>
</dependency>

和Hadoop 1.2.1，如下所示：

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>1.2.1</version>
</dependency>

问题是，我不确定为Spark配置hadoop的位置，以及配置hadoop conf的位置。因此，我可能添加了错误的Hadoop安装。另外，修改文件后是否需要重新启动？据我所知，我的机器上没有运行Hadoop服务。

我试图使用python文件的zip的GCS uri提交作业(通过--py-files参数),并将python文件名作为PY_FILE参数值.这似乎不起作用.我是否需要为PY_FILE值提供一些相对路径？PY_FILE也包含在zip中.例如in

gcloud beta dataproc jobs submit pyspark  --cluster clustername --py-files gcsuriofzip PY_FILE    

PY_FILE的价值应该是多少？

我一直在遇到一些Spark LDA主题建模中的各种问题(主要是看似随机间隔的解除错误)我一直在运行,我认为这主要与我的执行器上的内存分配不足有关.这似乎与有问题的自动群集配置有关.我的最新尝试使用n1-standard-8机器(8核,30GB RAM)用于主节点和工作节点(6个工作站,因此总共48个核心).

但是当我看到时,/etc/spark/conf/spark-defaults.conf我看到了这个:

spark.master yarn-client
spark.eventLog.enabled true
spark.eventLog.dir hdfs://cluster-3-m/user/spark/eventlog

# Dynamic allocation on YARN
spark.dynamicAllocation.enabled true
spark.dynamicAllocation.minExecutors 1
spark.dynamicAllocation.initialExecutors 100000
spark.dynamicAllocation.maxExecutors 100000
spark.shuffle.service.enabled true
spark.scheduler.minRegisteredResourcesRatio 0.0

spark.yarn.historyServer.address cluster-3-m:18080
spark.history.fs.logDirectory hdfs://cluster-3-m/user/spark/eventlog

spark.executor.cores 4
spark.executor.memory 9310m
spark.yarn.executor.memoryOverhead 930

# Overkill
spark.yarn.am.memory 9310m
spark.yarn.am.memoryOverhead 930

spark.driver.memory 7556m
spark.driver.maxResultSize 3778m
spark.akka.frameSize 512

# Add ALPN for Bigtable
spark.driver.extraJavaOptions -Xbootclasspath/p:/usr/local/share/google/alpn/alpn-boot-8.1.3.v20150130.jar
spark.executor.extraJavaOptions -Xbootclasspath/p:/usr/local/share/google/alpn/alpn-boot-8.1.3.v20150130.jar

但这些价值观没有多大意义.为什么只使用4/8执行器核心？并且只有9.3/30GB RAM？我的印象是所有这些配置都应该自动处理,但即使是我手动调整的尝试也没有让我到处都是.

例如,我尝试使用以下命令启动shell:

spark-shell --conf spark.executor.cores=8 --conf spark.executor.memory=24g

但后来失败了

java.lang.IllegalArgumentException: Required executor memory (24576+930 MB) is above the max threshold …

我正在Google DataProc群集上运行一个火花工作.但看起来Spark并没有使用vcores群集中的所有可用内容,如下所示

基于像其他一些问题,这个和这个,我已经安装使用群集DominantResourceCalculator来考虑资源分配的两个vCPU和内存

gcloud dataproc clusters create cluster_name --bucket="profiling-
job-default" \
--zone=europe-west1-c \
--master-boot-disk-size=500GB \
--worker-boot-disk-size=500GB \
--master-machine-type=n1-standard-16 \
--num-workers=10 \
--worker-machine-type=n1-standard-16 \
--initialization-actions gs://custom_init_gcp.sh \
--metadata MINICONDA_VARIANT=2 \
--properties=^--^yarn:yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator

但是,当我使用自定义火花标记提交我的作业时,看起来YARN不尊重这些自定义参数,默认使用内存作为资源计算的标准

gcloud dataproc jobs submit pyspark --cluster cluster_name \
--properties spark.sql.broadcastTimeout=900,spark.network.timeout=800\
,yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator\
,spark.dynamicAllocation.enabled=true\
,spark.executor.instances=10\
,spark.executor.cores=14\
,spark.executor.memory=15g\
,spark.driver.memory=50g \
src/my_python_file.py 

可以帮助有人弄清楚这里发生了什么？

通过添加以下内容，我可以很好地连接到驱动程序：

spark.driver.extraJavaOptions=-Dcom.sun.management.jmxremote \
                              -Dcom.sun.management.jmxremote.port=9178 \
                              -Dcom.sun.management.jmxremote.authenticate=false \
                              -Dcom.sun.management.jmxremote.ssl=false

但是做...

spark.executor.extraJavaOptions=-Dcom.sun.management.jmxremote \
                                -Dcom.sun.management.jmxremote.port=9178 \
                                -Dcom.sun.management.jmxremote.authenticate=false \
                                -Dcom.sun.management.jmxremote.ssl=false

...只会在驱动程序上产生一堆错误......

Container id: container_1501548048292_0024_01_000003
Exit code: 1
Stack trace: ExitCodeException exitCode=1: 
    at org.apache.hadoop.util.Shell.runCommand(Shell.java:972)
    at org.apache.hadoop.util.Shell.run(Shell.java:869)
    at org.apache.hadoop.util.Shell$ShellCommandExecutor.execute(Shell.java:1170)
    at org.apache.hadoop.yarn.server.nodemanager.DefaultContainerExecutor.launchContainer(DefaultContainerExecutor.java:236)
    at org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:305)
    at org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch.call(ContainerLaunch.java:84)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)


Container exited with a non-zero exit code 1

...最终使工作崩溃。

工人没有错误，它只是退出：

[org.apache.spark.util.ShutdownHookManager] - Shutdown hook called

Spark v2.2.0，集群是一个简单的 1m-2w 配置，我的作业在没有执行程序参数的情况下运行没有问题。

我正在尝试在 Dataproc 集群上提交 Spark 作业。该作业需要多个系统属性。我只能通过一个，如下所示：

gcloud dataproc jobs submit spark \                                   
    --cluster <cluster_name> \
    --class <class_name> \
    --properties spark.driver.extraJavaOptions=-Dhost=127.0.0.1  \
    --jars spark_job.jar

如何传递多个属性？我尝试如下，即使这样也不起作用。

--properties ^#^spark.driver.extraJavaOptions=-Dhost=127.0.0.1,-Dlimit=10

我正在使用 Google Cloud Dataproc 来做 spark 工作，我的编辑器是 Zepplin。我试图将 json 数据写入 gcp 存储桶。当我尝试 10MB 文件时，它成功了。但失败了 10GB 文件。我的 dataproc 有 1 个带有 4CPU、26GB 内存、500GB 磁盘的主服务器。5 名工人具有相同的配置。我想它应该能够处理 10GB 的数据。

我的命令是 toDatabase.repartition(10).write.json("gs://mypath")

错误是

org.apache.spark.SparkException: Job aborted.
  at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:224)
  at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:154)
  at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult$lzycompute(commands.scala:104)
  at org.apache.spark.sql.execution.command.DataWritingCommandExec.sideEffectResult(commands.scala:102)
  at org.apache.spark.sql.execution.command.DataWritingCommandExec.doExecute(commands.scala:122)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:656)
  at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:656)
  at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
  at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:656)
  at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:273)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:267)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:225)
  at org.apache.spark.sql.DataFrameWriter.json(DataFrameWriter.scala:528)
  ... 54 elided
Caused by: org.apache.spark.SparkException: …

我pyspark使用默认设置在8节点Google dataproc群集上运行.启动后几秒钟我看到30个执行器核心正在运行(如预期的那样):

    >>> sc.defaultParallelism
    30

一分钟后:

    >>> sc.defaultParallelism
    2

从那时起,所有操作仅在2个核心上运行:

    >>> rng = sc.parallelize(range(1,1000000))
    >>> rng.cache()
    >>> rng.count()
    >>> rng.getNumPartitions()
    2

如果我rng.cache()在核心仍处于连接状态时运行,则它们会保持连接并且作业会分配

检查监控应用程序(主节点上的端口4040)显示执行程序已删除:

Executor 1
Removed at 2016/02/25 16:20:14
Reason: Container container_1456414665542_0006_01_000002 exited from explicit termination request." 

是否有一些设置可以保持核心连接而无需解决方法？

我有一份工作需要很长时间才能在DataProc上运行.与此同时,我需要能够运行其他较小的工作.

根据我从Google Dataproc文档中收集的内容,该平台应该支持多个作业,因为它使用YARN动态分配资源.

但是,当我尝试启动多个作业时,它们会排队,并且在群集空闲之前不会启动.

默认情况下,所有设置.如何启用同时运行的多个作业？

我使用以下命令创建一个dataproc集群

gcloud dataproc clusters create datascience \
--initialization-actions \
    gs://dataproc-initialization-actions/jupyter/jupyter.sh \

但是，当我提交我的PySpark作业时，出现以下错误

例外：worker中的Python版本与驱动程序3.7中的版本不同，PySpark无法以其他次要版本运行。请检查环境变量PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON是否已正确设置。

有什么想法吗？