如何确保Apache Spark安装代码只运行一次？

Question

我正在Scala中编写一个Spark作业，它读取S3上的parquet文件，执行一些简单的转换，然后将它们保存到DynamoDB实例中。每次运行时，我们都需要在Dynamo中创建一个新的表，所以我编写了一个Lambda函数，它负责表的创建。Spark作业所做的第一件事是生成一个表名，调用我的Lambda函数（将新表名传递给它），等待创建表，然后正常地执行ETL步骤。

但是，看起来我的Lambda函数总是被调用两次。我无法解释。下面是代码的示例：

def main(spark: SparkSession, pathToParquet: String) {

  // generate a unique table name
  val tableName = generateTableName()

  // call the lambda function
  val result = callLambdaFunction(tableName)

  // wait for the table to be created
  waitForTableCreation(tableName)

  // normal ETL pipeline
  var parquetRDD = spark.read.parquet(pathToParquet)
  val transformedRDD = parquetRDD.map((row: Row) => transformData(row), encoder=kryo[(Text, DynamoDBItemWritable)])
  transformedRDD.saveAsHadoopDataset(getConfiguration(tableName))
  spark.sparkContext.stop()
}

def waitForTableCreation(tableName: String) {
  val client: AmazonDynamoDB = AmazonDynamoDBClientBuilder.defaultClient()
  val waiter: Waiter[DescribeTableRequest] = client.waiters().tableExists()
  try {
    waiter.run(new WaiterParameters[DescribeTableRequest](new DescribeTableRequest(tableName)))
  } catch {
      case ex: WaiterTimedOutException =>
        LOGGER.error("Timed out waiting to create table: " + tableName)
        throw ex
      case t: Throwable => throw t
  }
}

lambda调用同样简单：

def callLambdaFunction(tableName: String) {
  val myLambda = LambdaInvokerFactory.builder()
    .lambdaClient(AWSLambdaClientBuilder.defaultClient)
    .lambdaFunctionNameResolver(new LambdaByName(LAMBDA_FUNCTION_NAME))
    .build(classOf[MyLambdaContract])
  myLambda.invoke(new MyLambdaInput(tableName))
}

正如我所说，当我在这段代码上运行spark-submit时，它肯定会碰到Lambda函数。但我无法解释为什么它会击中它两次。结果是在DynamoDB中提供了两个表。

在将此作为Spark作业运行的上下文中，等待步骤似乎也失败了。但是当我对等待的代码进行单元测试时，它似乎可以自己正常工作。它成功地阻止，直到表准备就绪。

下面是我的getconfiguration函数的代码：

def getConfiguration(tableName: String) : JobConf = {
  val conf = new Configuration()
  conf.set("dynamodb.servicename", "dynamodb")
  conf.set("dynamodb.input.tableName", tableName)
  conf.set("dynamodb.output.tableName", tableName)
  conf.set("dynamodb.endpoint", "https://dynamodb.us-east-1.amazonaws.com")
  conf.set("dynamodb.regionid", "us-east-1")
  conf.set("mapred.output.format.class", "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
  conf.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")
  new JobConf(conf)
}

另外，这里有一个要点，其中包含了一些我在尝试运行此代码时看到的异常日志。

Answer 1

感谢@soapergem添加日志记录和选项。我添加了一个答案（一个尝试），因为它可能比注释长一点：）

总结一下：

• spark-submit和配置选项没有什么奇怪的地方
• 在https://gist.github.com/soapergem/6b379b5a9092dcd43777bdec8dee65a8#file-stderr-log中，可以看到应用程序执行了两次。它两次从接受状态转到运行状态。这与EMR默认值一致（如何防止EMR火花步骤重试？）。为了确认这一点，您可以检查在执行该步骤后是否创建了2个表（我想这里是使用动态名称生成表；每次执行时使用不同的名称，如果重试，应该提供2个不同的名称）

Caused by: com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException: Requested resource not found: Table: EmrTest_20190708143902 not found (Service: AmazonDynamoDBv2; Status Code: 400; Error Code: ResourceNotFoundException; Request ID: V0M91J7KEUVR4VM78MF5TKHLEBVV4KQNSO5AEMVJF66Q9ASUAAJG)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1712)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1367)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeHelper(AmazonHttpClient.java:1113)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.doExecute(AmazonHttpClient.java:770)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeWithTimer(AmazonHttpClient.java:744)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.execute(AmazonHttpClient.java:726)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutor.access$500(AmazonHttpClient.java:686)
    at com.amazonaws.http.AmazonHttpClient$RequestExecutionBuilderImpl.execute(AmazonHttpClient.java:668)
    at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:532)
    at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:512)
    at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.doInvoke(AmazonDynamoDBClient.java:4243)
    at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.invoke(AmazonDynamoDBClient.java:4210)
    at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.executeDescribeTable(AmazonDynamoDBClient.java:1890)
    at com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient.describeTable(AmazonDynamoDBClient.java:1857)
    at org.apache.hadoop.dynamodb.DynamoDBClient$1.call(DynamoDBClient.java:129)
    at org.apache.hadoop.dynamodb.DynamoDBClient$1.call(DynamoDBClient.java:126)
at org.apache.hadoop.dynamodb.DynamoDBFibonacciRetryer.runWithRetry(DynamoDBFibonacciRetryer.java:80)

• 对于第一次执行，请确保您拥有所有权限。IMO将是dynamodb:describe*关于资源:*（如果是这个原因，为了最小特权原则，您应该在生产中使用somthing资源:test_emr*)
• 第二次执行时，删除dynamodb:descripbe*并检查是否得到与gist中相同的堆栈跟踪