从AWS s3存储桶中读取地板数据
问题描述:
我需要从aws s3读取地板数据。如果我使用AWS SDK为了这个,我能得到的InputStream这样的:从AWS s3存储桶中读取地板数据
S3Object object = s3Client.getObject(new GetObjectRequest(bucketName, bucketKey));
InputStream inputStream = object.getObjectContent();
但Apache的实木复合地板读者仅使用本地文件是这样的:
ParquetReader<Group> reader =
ParquetReader.builder(new GroupReadSupport(), new Path(file.getAbsolutePath()))
.withConf(conf)
.build();
reader.read()
所以我不知道如何解析输入流为镶木地板文件。 例如,对于csv文件,存在使用输入流的CSVParser。
我知道解决方案使用火花这个目标。 像这样:
SparkSession spark = SparkSession
.builder()
.getOrCreate();
Dataset<Row> ds = spark.read().parquet("s3a://bucketName/file.parquet");
但我不能使用的火花。
任何人都可以告诉我任何解决方案从S3阅读实木复合地板数据?
答
String SCHEMA_TEMPLATE = "{" +
"\"type\": \"record\",\n" +
" \"name\": \"schema\",\n" +
" \"fields\": [\n" +
" {\"name\": \"timeStamp\", \"type\": \"string\"},\n" +
" {\"name\": \"temperature\", \"type\": \"double\"},\n" +
" {\"name\": \"pressure\", \"type\": \"double\"}\n" +
" ]" +
"}";
String PATH_SCHEMA = "s3a";
Path internalPath = new Path(PATH_SCHEMA, bucketName, folderName);
Schema schema = new Schema.Parser().parse(SCHEMA_TEMPLATE);
Configuration configuration = new Configuration();
AvroReadSupport.setRequestedProjection(configuration, schema);
ParquetReader<GenericRecord> = AvroParquetReader.GenericRecord>builder(internalPath).withConf(configuration).build();
GenericRecord genericRecord = parquetReader.read();
while(genericRecord != null) {
Map<String, String> valuesMap = new HashMap<>();
genericRecord.getSchema().getFields().forEach(field -> valuesMap.put(field.name(), genericRecord.get(field.name()).toString()));
genericRecord = parquetReader.read();
}
摇篮相依
compile 'com.amazonaws:aws-java-sdk:1.11.213'
compile 'org.apache.parquet:parquet-avro:1.9.0'
compile 'org.apache.parquet:parquet-hadoop:1.9.0'
compile 'org.apache.hadoop:hadoop-common:2.8.1'
compile 'org.apache.hadoop:hadoop-aws:2.8.1'
compile 'org.apache.hadoop:hadoop-client:2.8.1'