hadoop - Spark/Hadoop/Yarn集群通信需要外部ip？

Question

我使用 bdutil 在 Hadoop (2.6) 集群上部署了带有 yarn-client 的 Spark (1.3.1)，默认情况下，实例是使用临时外部 ips 创建的，到目前为止 spark 工作正常。考虑到一些安全问题，并假设集群只能在内部访问，我从实例中删除了外部 ips；在那之后，spark-shell 甚至不会运行，并且似乎无法与 Yarn/Hadoop 通信，并且会无限期地卡住。只有在我重新添加外部 ips 之后，spark-shell 才开始正常工作。

我的问题是，运行 spark over yarn 是否需要节点的外部 ips，为什么？如果是，是否会有关于安全等方面的担忧？谢谢!

Answer 1

简答题

您需要外部 IP 地址才能访问 GCS，默认的 bdutil 设置将 GCS 设置为默认的 Hadoop 文件系统，包括控制文件。使用 ./bdutil -F hdfs ... deploy 改为使用 HDFS 作为默认值。

除非您在 GCE 网络配置中的防火墙规则中添加了过多的许可规则，否则在使用外部 IP 地址时不应担心安全问题。

编辑:目前似乎存在一个错误，我们将 spark.eventLog.dir 设置为 GCS 路径，即使 default_fs 是 hdfs。我提交了 https://github.com/GoogleCloudPlatform/bdutil/issues/35跟踪这个。在此期间，只需在您的主机上手动编辑 /home/hadoop/spark-install/conf/spark-defaults.conf(您可能需要 sudo -u hadoop vim.tiny/home/hadoop/spark-install/conf/spark-defaults.conf 对其具有编辑权限)将 spark.eventLog.dir 设置为 hdfs:///spark-eventlog -base 或 HDFS 中的其他内容，然后运行 ​​hadoop fs -mkdir -p hdfs:///spark-eventlog-base 使其正常工作。

长答案

默认情况下，bdutil 还将 Google Cloud Storage 配置为“默认 Hadoop 文件系统”，这意味着 Spark 和 YARN 使用的控制文件需要访问 Google Cloud Storage。此外，还需要外部 IP 才能访问 Google Cloud Storage。

在手动配置网络内 SSH 后，我确实设法部分重现了您的案例；在启动过程中，我实际上看到了以下内容:

15/06/26 17:23:05 INFO yarn.Client: Preparing resources for our AM container
15/06/26 17:23:05 INFO gcs.GoogleHadoopFileSystemBase: GHFS version: 1.4.0-hadoop2
15/06/26 17:23:26 WARN http.HttpTransport: exception thrown while executing request
java.net.SocketTimeoutException: connect timed out
  at java.net.PlainSocketImpl.socketConnect(Native Method)
  at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:339)
  at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:200)
  at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:182)
  at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
  at java.net.Socket.connect(Socket.java:579)
  at sun.security.ssl.SSLSocketImpl.connect(SSLSocketImpl.java:625)
  at sun.net.NetworkClient.doConnect(NetworkClient.java:175)
  at sun.net.www.http.HttpClient.openServer(HttpClient.java:432)
  at sun.net.www.http.HttpClient.openServer(HttpClient.java:527)
  at sun.net.www.protocol.https.HttpsClient.<init>(HttpsClient.java:275)
  at sun.net.www.protocol.https.HttpsClient.New(HttpsClient.java:371)
  at sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.getNewHttpClient(AbstractDelegateHttpsURLConnection.java:191)
  at sun.net.www.protocol.http.HttpURLConnection.plainConnect(HttpURLConnection.java:933)
  at sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.connect(AbstractDelegateHttpsURLConnection.java:177)
  at sun.net.www.protocol.https.HttpsURLConnectionImpl.connect(HttpsURLConnectionImpl.java:153)
  at com.google.api.client.http.javanet.NetHttpRequest.execute(NetHttpRequest.java:93)
  at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:965)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:410)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:343)
  at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:460)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.getBucket(GoogleCloudStorageImpl.java:1557)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl.getItemInfo(GoogleCloudStorageImpl.java:1512)
  at com.google.cloud.hadoop.gcsio.CacheSupplementedGoogleCloudStorage.getItemInfo(CacheSupplementedGoogleCloudStorage.java:516)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageFileSystem.getFileInfo(GoogleCloudStorageFileSystem.java:1016)
  at com.google.cloud.hadoop.gcsio.GoogleCloudStorageFileSystem.exists(GoogleCloudStorageFileSystem.java:382)
  at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystemBase.configureBuckets(GoogleHadoopFileSystemBase.java:1639)
  at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem.configureBuckets(GoogleHadoopFileSystem.java:71)
  at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystemBase.configure(GoogleHadoopFileSystemBase.java:1587)
  at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystemBase.initialize(GoogleHadoopFileSystemBase.java:776)
  at com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystemBase.initialize(GoogleHadoopFileSystemBase.java:739)
  at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2596)
  at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
  at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2630)
  at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2612)
  at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:370)
  at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:169)
  at org.apache.spark.deploy.yarn.Client.prepareLocalResources(Client.scala:216)
  at org.apache.spark.deploy.yarn.Client.createContainerLaunchContext(Client.scala:384)
  at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:102)
  at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:58)
  at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:141)
  at org.apache.spark.SparkContext.<init>(SparkContext.scala:381)
  at org.apache.spark.repl.SparkILoop.createSparkContext(SparkILoop.scala:1016)
  at $line3.$read$$iwC$$iwC.<init>(<console>:9)
  at $line3.$read$$iwC.<init>(<console>:18)
  at $line3.$read.<init>(<console>:20)
  at $line3.$read$.<init>(<console>:24)
  at $line3.$read$.<clinit>(<console>)
  at $line3.$eval$.<init>(<console>:7)
  at $line3.$eval$.<clinit>(<console>)
  at $line3.$eval.$print(<console>)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:606)
  at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
  at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1338)
  at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
  at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
  at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
  at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:856)
  at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:901)
  at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:813)
  at org.apache.spark.repl.SparkILoopInit$$anonfun$initializeSpark$1.apply(SparkILoopInit.scala:123)
  at org.apache.spark.repl.SparkILoopInit$$anonfun$initializeSpark$1.apply(SparkILoopInit.scala:122)
  at org.apache.spark.repl.SparkIMain.beQuietDuring(SparkIMain.scala:324)
  at org.apache.spark.repl.SparkILoopInit$class.initializeSpark(SparkILoopInit.scala:122)
  at org.apache.spark.repl.SparkILoop.initializeSpark(SparkILoop.scala:64)
  at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1$$anonfun$apply$mcZ$sp$5.apply$mcV$sp(SparkILoop.scala:973)
  at org.apache.spark.repl.SparkILoopInit$class.runThunks(SparkILoopInit.scala:157)
  at org.apache.spark.repl.SparkILoop.runThunks(SparkILoop.scala:64)
  at org.apache.spark.repl.SparkILoopInit$class.postInitialization(SparkILoopInit.scala:106)
  at org.apache.spark.repl.SparkILoop.postInitialization(SparkILoop.scala:64)
  at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply$mcZ$sp(SparkILoop.scala:990)
  at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:944)
  at org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:944)
  at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
  at org.apache.spark.repl.SparkILoop.org$apache$spark$repl$SparkILoop$$process(SparkILoop.scala:944)
  at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:1058)
  at org.apache.spark.repl.Main$.main(Main.scala:31)
  at org.apache.spark.repl.Main.main(Main.scala)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:606)
  at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:569)
  at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:166)
  at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:189)
  at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:110)
  at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

正如预期的那样，只需调用 org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start 它就会尝试联系 Google Cloud Storage，但失败了，因为没有外部 IP 就无法访问 GCS。

为了解决这个问题，您可以在创建集群时简单地使用 -F hdfs 以使用 HDFS 作为默认文件系统；在那种情况下，即使没有外部 IP 地址，一切都应该在集群内工作。在该模式下，您甚至可以在通过指定完整的 gs://bucket/object 路径作为 Hadoop 参数分配外部 IP 地址时继续使用 GCS。但是请注意，在这种情况下，只要您删除了外部 IP 地址，就无法使用 GCS，除非您还配置了代理服务器并通过代理传输所有数据； GCS 配置为 fs.gs.proxy.address。

一般来说，除非您在 Google Compute Engine 的“默认”网络防火墙规则中开启了新的许可规则，否则无需因为拥有外部 IP 地址而担心安全问题。