除了挖掘日志文件以获取日志输出之外，还有其他方法可以获取 Spark 跟踪 URL？

Question

我有一个创建 Spark 会话的 Scala 应用程序，并且我设置了使用 Spark REST API 的运行状况检查。 Spark 应用程序本身在 Hadoop Yarn 上运行。当前通过读取创建 Spark 会话时生成的 Spark 日志记录来检索 REST API URL。这在大多数情况下都有效，但在我的应用程序中有一些边缘情况不能很好地工作。 有谁知道获取此跟踪 URL 的另一种方法？

Answer 1

“您可以通过读取 YARN 配置中的 yarn.resourcemanager.webapp.address 值和应用程序 ID（在侦听器总线上发送的事件和现有的 SparkContext 方法中都公开）来做到这一点。”

根据开发者的回复复制上述段落：https://issues.apache.org/jira/browse/SPARK-20458

更新：

我确实尝试了解决方案并且非常接近。下面是一些用于构建该 URL 的 Scala/Spark 代码：

@transient val ssc: StreamingContext = StreamingContext.getActiveOrCreate(rabbitSettings.checkpointPath, CreateStreamingContext)

// Update yarn logs URL in Elasticsearch
YarnLogsTracker.update(
  ssc.sparkContext.uiWebUrl,
  ssc.sparkContext.applicationId,
  "test2")

YarnLogsTracker 对象是这样的：

object YarnLogsTracker {

  private def recoverURL(u: Option[String]): String = u match {
    case Some(a) => a.split(":").take(2).mkString(":")
    case None => ""
  }

 def update(rawUrl: Option[String], rawAppId: String, tenant: String): Unit = {
   val logUrl = s"${recoverURL(rawUrl)}:8042/node/containerlogs/container${rawAppId.substring(11)}_01_000002/$tenant/stdout/?start=-4096"
...

这会产生如下内容：http://10.99.25.146:8042/node/containerlogs/container_1516203096033_91164_01_000002/test2/stdout/?start=-4096

Answer 2

我发现了一种“合理”的方法来获得它。显然，最好的方法是让 Spark 库将它们已经直接获取到启动器应用程序的ApplicationReport 公开，因为它们会遇到设置委托令牌等的麻烦。但是，这似乎不太可能发生。

这种方法是双管齐下的。首先，它尝试自己构建一个YarnClient，以获取ApplicationReport，这将具有权威的跟踪URL。但是，根据我的经验，这可能会失败（例如：如果作业在 CLUSTER 模式下运行，并且在 Kerberized 环境中使用 --proxy-user，那么这将无法正确向 YARN 进行身份验证。

在我的例子中，我从驱动程序本身调用这个辅助方法，并将结果报告回我的启动器应用程序。但是，原则上，您拥有 Hadoop Configuration 的任何地方都应该可以工作（可能包括您的启动器应用程序）。显然，您可以根据您的需要和对复杂性、额外处理等的容忍度来使用此实现的“叉”（或两者）。

  /**
   * Given a Hadoop {@link org.apache.hadoop.conf.Configuration} and appId, use the YARN API (via an
   * {@link YarnClient} instance) to get the application report, which includes the trackingUrl.  If this fails,
   * then as a fallback, it attempts to "guess" the URL by looking at various YARN configuration properties,
   * and assumes that the URL will be something like: <pre>[yarnWebUI:port]/proxy/[appId]</pre>.
   *
   * @param hadoopConf the Hadoop {@link org.apache.hadoop.conf.Configuration}
   * @param appId the YARN application ID
   * @return the app trackingUrl, either retrieved using the {@link YarnClient}, or manually constructed using
   *         the fallback approach
   */
  public static String getYarnApplicationTrackingUrl(org.apache.hadoop.conf.Configuration hadoopConf, String appId) {
    LOG.debug("Attempting to look up YARN url for applicationId {}", appId);
    YarnClient yarnClient = null;
    try {
      // do not attempt to fail over on authentication error (ex: running with proxy-user and Kerberos)
      hadoopConf.set("yarn.client.failover-max-attempts", "0");
      yarnClient = YarnClient.createYarnClient();
      yarnClient.init(hadoopConf);
      yarnClient.start();

      final ApplicationReport report = yarnClient.getApplicationReport(ConverterUtils.toApplicationId(appId));
      return report.getTrackingUrl();
    } catch (YarnException | IOException e) {
      LOG.warn(
          "{} attempting to get report for YARN appId {}; attempting to use manually constructed fallback",
          e.getClass().getSimpleName(),
          appId,
          e
      );

      String baseYarnWebappUrl;
      String protocol;
      if ("HTTPS_ONLY".equals(hadoopConf.get("yarn.http.policy"))) {
        // YARN is configured to use HTTPS only, hence return the https address
        baseYarnWebappUrl = hadoopConf.get("yarn.resourcemanager.webapp.https.address");
        protocol = "https";
      } else {
        baseYarnWebappUrl = hadoopConf.get("yarn.resourcemanager.webapp.address");
        protocol = "http";
      }

      return String.format("%s://%s/proxy/%s", protocol, baseYarnWebappUrl, appId);
    } finally {
      if (yarnClient != null) {
        yarnClient.stop();
      }
    }
  }