EMR：如何将 Spark 与 Hive 集成？

Question

使用 EMR 集群，我创建了一个映射到 DynamoDB 表的外部 Hive 表（超过 8 亿行）。它运行良好，我可以通过 hive 进行查询和插入。

如果我尝试使用 Hive 中的 hash_key 条件进行查询，我会在几秒钟内得到结果。但是使用 SparkSQL 和 enableHiveSupport（访问 Hive）通过 spark-submit 执行相同的查询并没有完成。似乎从 Spark 它正在对表进行完整扫描。

我尝试了几种配置（例如不同的 hive-site.xml），但在 Spark 上似乎效果不佳。我应该如何通过 Spark 做到这一点？有什么建议吗？

谢谢

Answer 1

只要确保使用 AWS 开源的 dynamo 连接器即可。默认情况下，它在 EMR AFAIK 上可用。

使用 DynamoDBStorageHandler 类创建表的语法：

CREATE EXTERNAL TABLE hive_tablename (
    hive_column1_name column1_datatype,
    hive_column2_name column2_datatype
)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES (
    "dynamodb.table.name" = "dynamodb_tablename",
    "dynamodb.column.mapping" =
        "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name"
);

对于任何 Spark Job，您需要有以下配置：

$ spark-shell --jars /usr/share/aws/emr/ddb/lib/emr-ddb-hadoop.jar
...
import org.apache.hadoop.io.Text;
import org.apache.hadoop.dynamodb.DynamoDBItemWritable
import org.apache.hadoop.dynamodb.read.DynamoDBInputFormat
import org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat
import org.apache.hadoop.mapred.JobConf
import org.apache.hadoop.io.LongWritable

var jobConf = new JobConf(sc.hadoopConfiguration)
jobConf.set("dynamodb.input.tableName", "myDynamoDBTable")

jobConf.set("mapred.output.format.class", "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
jobConf.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")

var orders = sc.hadoopRDD(jobConf, classOf[DynamoDBInputFormat], classOf[Text], classOf[DynamoDBItemWritable])

orders.count()

参考资料： https://github.com/awslabs/emr-dynamodb-connector