在包含json数据的parquet文件上创建配置单元表

jgwigjjp 于 2021-05-29 发布在 Hadoop

关注(0)|答案(1)|浏览(377)

我要达到的目标
从源大json文件（employee sample.json）获取数据
一个简单的spark应用程序，将其作为文本文件读取并存储在parquet（simple-loader.java）中。我不知道json文件中有什么，所以我不能放任何模式，所以我希望模式是读的，而不是写的。创建了一个parquet文件，其中一列名为“value”，包含json字符串
在parquet文件上创建一个hive外部表，当我执行“select*fromtable”时，我看到一列带有json数据。
我真正需要的是创建一个hive表，它可以读取“value”列中的json数据并应用schema和emit列，这样我就可以根据需要在原始数据上创建各种表。
我已经在json文件上创建了配置单元表，并提取了列，但是这个从parquet提取列并应用json模式欺骗了我

employee-sample.json
{"name":"Dave", "age" : 30 , "DOB":"1987-01-01"}
{"name":"Steve", "age" : 31 , "DOB":"1986-01-01"}
{"name":"Kumar", "age" : 32 , "DOB":"1985-01-01"}

将json转换为parquet的简单spark代码

simple-loader.java
public static void main(String[] args) {
    SparkSession sparkSession = SparkSession.builder()
            .appName(JsonToParquet.class.getName())
            .master("local[*]").getOrCreate();
    Dataset<String> eventsDataSet = sparkSession.read().textFile("D:\\dev\\employee-sample.json");
    eventsDataSet.createOrReplaceTempView("rawView");
    sparkSession.sqlContext().sql("select string(value) as value from rawView")
            .write()
            .parquet("D:\\dev\\" + UUID.randomUUID().toString());
    sparkSession.close();
}

Parquet文件上的Hive桌

CREATE EXTERNAL TABLE EVENTS_RAW (
VALUE STRING)
STORED AS PARQUET
LOCATION 'hdfs://XXXXXX:8020/employee/data_raw';

我尝试设置json serde，但它只在数据以json foram存储时有效，行格式serde'com.proofpoint.hive.serde.jsonserde'
预期格式

CREATE EXTERNAL TABLE EVENTS_DATA (
    NAME STRING,
    AGE STRING,
    DOB STRING)
??????????????????????????????

hadoop Hive apache-spark parquet hiveql

来源：https://stackoverflow.com/questions/47818707/creating-hive-table-on-parquet-file-which-has-json-data

1条答案

按热度按时间

q5iwbnjs1#

创建配置单元外部表示例：

public static final String CREATE_EXTERNAL = "CREATE EXTERNAL TABLE %s" +
        " (%s) " +
        " PARTITIONED BY(%s) " +
        " STORED AS %s" +
        " LOCATION '%s'";
/**
 * Will create an external table and recover the partitions
 */
public void createExternalTable(SparkSession sparkSession, StructType schema, String tableName, SparkFormat format, List<StructField> partitions, String tablePath){
    String createQuery = createTableString(schema, tableName, format, partitions, tablePath);
    logger.info("Going to create External table with the following query:\n " + createQuery);
    sparkSession.sql(createQuery);
    logger.debug("Finish to create External table with the following query:\n " + createQuery);
    recoverPartitions(sparkSession, tableName);
}
public String createTableString(StructType schema, String tableName, SparkFormat format, List<StructField> partitions, String tablePath){
    Set<String> partitionNames = partitions.stream().map(struct -> struct.name()).collect(Collectors.toSet());
    String columns = Arrays.stream(schema.fields())
            //Filter the partitions
            .filter(field -> !partitionNames.contains(field.name()))
            //
            .map(HiveTableHelper::fieldToStringBuilder)
            .collect(Collectors.joining(", "));
    String partitionsString = partitions.stream().map(HiveTableHelper::fieldToStringBuilder).collect(Collectors.joining(", "));
    return String.format(CREATE_EXTERNAL, tableName, columns, partitionsString, format.name(), tablePath);
}
/**
 *
 * @param sparkSession
 * @param table
 */
public void recoverPartitions(SparkSession sparkSession, String table){
    String query = "ALTER TABLE " + table + " RECOVER PARTITIONS";
    logger.debug("Start: " + query);
    sparkSession.sql(query);
    sparkSession.catalog().refreshTable(table);
    logger.debug("Finish: " + query);
}
  public static StringBuilder fieldToStringBuilder(StructField field){
    StringBuilder sb = new StringBuilder();
    sb.append(field.name()).append( " ").append(field.dataType().simpleString());
    return sb;
}

展开查看全部

赞(0）回复(0）举报 2021-05-29

我来回答

在包含json数据的parquet文件上创建配置单元表

1条答案

相关问题

热门标签

最新问答