You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2019/08/23 09:22:24 UTC
[incubator-pinot] 01/01: Support aws s3 with Parquet in pinot-tools
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch support_aws_in_hadoop
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 636204c06df3c5d563cc2fa01035d80cea22a243
Author: Xiang Fu <fx...@gmail.com>
AuthorDate: Fri Aug 23 02:22:05 2019 -0700
Support aws s3 with Parquet in pinot-tools
---
.../apache/pinot/core/data/readers/FileFormat.java | 2 +-
pinot-tools/pom.xml | 14 ++++
.../tools/admin/command/CreateSegmentCommand.java | 91 +++++++++++++++-------
pom.xml | 5 ++
4 files changed, 82 insertions(+), 30 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/FileFormat.java b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/FileFormat.java
index 5f7120d..43f8391 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/readers/FileFormat.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/readers/FileFormat.java
@@ -19,5 +19,5 @@
package org.apache.pinot.core.data.readers;
public enum FileFormat {
- AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, OTHER
+ AVRO, GZIPPED_AVRO, CSV, JSON, PINOT, THRIFT, PARQUET, OTHER
}
diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml
index 55c2ecd..999f034 100644
--- a/pinot-tools/pom.xml
+++ b/pinot-tools/pom.xml
@@ -56,6 +56,11 @@
</dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-parquet</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
<artifactId>pinot-connector-kafka-base</artifactId>
<version>${project.version}</version>
</dependency>
@@ -98,6 +103,15 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-math3</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-aws</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/CreateSegmentCommand.java b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/CreateSegmentCommand.java
index 95c1371..69e11cc 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/CreateSegmentCommand.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/command/CreateSegmentCommand.java
@@ -19,20 +19,29 @@
package org.apache.pinot.tools.admin.command;
import java.io.File;
-import java.io.FilenameFilter;
import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
+import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.pinot.common.data.StarTreeIndexSpec;
import org.apache.pinot.common.utils.JsonUtils;
import org.apache.pinot.core.data.readers.FileFormat;
+import org.apache.pinot.core.data.readers.RecordReader;
+import org.apache.pinot.core.data.readers.RecordReaderFactory;
import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
import org.apache.pinot.core.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.parquet.data.readers.ParquetRecordReader;
import org.apache.pinot.startree.hll.HllConfig;
import org.apache.pinot.startree.hll.HllConstants;
import org.apache.pinot.tools.Command;
@@ -47,40 +56,28 @@ import org.slf4j.LoggerFactory;
*/
public class CreateSegmentCommand extends AbstractBaseAdminCommand implements Command {
private static final Logger LOGGER = LoggerFactory.getLogger(CreateSegmentCommand.class);
-
+ @Option(name = "-enableStarTreeIndex", usage = "Enable Star Tree Index.")
+ boolean _enableStarTreeIndex = false;
@Option(name = "-generatorConfigFile", metaVar = "<string>", usage = "Config file for segment generator.")
private String _generatorConfigFile;
-
@Option(name = "-dataDir", metaVar = "<string>", usage = "Directory containing the data.")
private String _dataDir;
-
@Option(name = "-format", metaVar = "<AVRO/CSV/JSON>", usage = "Input data format.")
private FileFormat _format;
-
@Option(name = "-outDir", metaVar = "<string>", usage = "Name of output directory.")
private String _outDir;
-
@Option(name = "-overwrite", usage = "Overwrite existing output directory.")
private boolean _overwrite = false;
-
@Option(name = "-tableName", metaVar = "<string>", usage = "Name of the table.")
private String _tableName;
-
@Option(name = "-segmentName", metaVar = "<string>", usage = "Name of the segment.")
private String _segmentName;
-
@Option(name = "-timeColumnName", metaVar = "<string>", usage = "Primary time column.")
private String _timeColumnName;
-
@Option(name = "-schemaFile", metaVar = "<string>", usage = "File containing schema for data.")
private String _schemaFile;
-
@Option(name = "-readerConfigFile", metaVar = "<string>", usage = "Config file for record reader.")
private String _readerConfigFile;
-
- @Option(name = "-enableStarTreeIndex", usage = "Enable Star Tree Index.")
- boolean _enableStarTreeIndex = false;
-
@Option(name = "-starTreeIndexSpecFile", metaVar = "<string>", usage = "Config file for star tree index.")
private String _starTreeIndexSpecFile;
@@ -283,24 +280,22 @@ public class CreateSegmentCommand extends AbstractBaseAdminCommand implements Co
}
// Filter out all input files.
- File dir = new File(_dataDir);
- if (!dir.exists() || !dir.isDirectory()) {
+ final Path dataDirPath = new Path(_dataDir);
+ FileSystem fileSystem = FileSystem.get(URI.create(_dataDir), new Configuration());
+
+ if (!fileSystem.exists(dataDirPath) || !fileSystem.isDirectory(dataDirPath)) {
throw new RuntimeException("Data directory " + _dataDir + " not found.");
}
- File[] files = dir.listFiles(new FilenameFilter() {
- @Override
- public boolean accept(File dir, String name) {
- return name.toLowerCase().endsWith(_format.toString().toLowerCase());
- }
- });
+ // Gather all data files
+ List<Path> dataFilePaths = getDataFilePaths(dataDirPath);
- if ((files == null) || (files.length == 0)) {
+ if ((dataFilePaths == null) || (dataFilePaths.size() == 0)) {
throw new RuntimeException(
"Data directory " + _dataDir + " does not contain " + _format.toString().toUpperCase() + " files.");
}
- LOGGER.info("Accepted files: {}", Arrays.toString(files));
+ LOGGER.info("Accepted files: {}", Arrays.toString(dataFilePaths.toArray()));
// Make sure output directory does not already exist, or can be overwritten.
File outDir = new File(_outDir);
@@ -362,7 +357,7 @@ public class CreateSegmentCommand extends AbstractBaseAdminCommand implements Co
ExecutorService executor = Executors.newFixedThreadPool(_numThreads);
int cnt = 0;
- for (final File file : files) {
+ for (final Path dataFilePath : dataFilePaths) {
final int segCnt = cnt;
executor.execute(new Runnable() {
@@ -370,12 +365,23 @@ public class CreateSegmentCommand extends AbstractBaseAdminCommand implements Co
public void run() {
try {
SegmentGeneratorConfig config = new SegmentGeneratorConfig(segmentGeneratorConfig);
- config.setInputFilePath(file.getAbsolutePath());
+
+ String localFile = dataFilePath.getName();
+ Path localFilePath = new Path(localFile);
+ dataDirPath.getFileSystem(new Configuration()).copyToLocalFile(dataFilePath, localFilePath);
+ config.setInputFilePath(localFile);
config.setSegmentName(_segmentName + "_" + segCnt);
config.loadConfigFiles();
-
final SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
- driver.init(config);
+ switch (config.getFormat()) {
+ case PARQUET:
+ RecordReader parquetRecordReader = new ParquetRecordReader();
+ parquetRecordReader.init(config);
+ driver.init(config, parquetRecordReader);
+ break;
+ default:
+ driver.init(config);
+ }
driver.build();
} catch (Exception e) {
throw new RuntimeException(e);
@@ -388,4 +394,31 @@ public class CreateSegmentCommand extends AbstractBaseAdminCommand implements Co
executor.shutdown();
return executor.awaitTermination(1, TimeUnit.HOURS);
}
+
+ protected List<Path> getDataFilePaths(Path pathPattern)
+ throws IOException {
+ List<Path> tarFilePaths = new ArrayList<>();
+ FileSystem fileSystem = FileSystem.get(pathPattern.toUri(), new Configuration());
+ getDataFilePathsHelper(fileSystem, fileSystem.globStatus(pathPattern), tarFilePaths);
+ return tarFilePaths;
+ }
+
+ protected void getDataFilePathsHelper(FileSystem fileSystem, FileStatus[] fileStatuses, List<Path> tarFilePaths)
+ throws IOException {
+ for (FileStatus fileStatus : fileStatuses) {
+ Path path = fileStatus.getPath();
+ if (fileStatus.isDirectory()) {
+ getDataFilePathsHelper(fileSystem, fileSystem.listStatus(path), tarFilePaths);
+ } else {
+ if (isDataFile(path.getName())) {
+ tarFilePaths.add(path);
+ }
+ }
+ }
+ }
+
+ protected boolean isDataFile(String fileName) {
+ return fileName.endsWith(".avro") || fileName.endsWith(".csv") || fileName.endsWith(".json") || fileName
+ .endsWith(".thrift") || fileName.endsWith(".parquet");
+ }
}
diff --git a/pom.xml b/pom.xml
index cc8767c..d770f29 100644
--- a/pom.xml
+++ b/pom.xml
@@ -657,6 +657,11 @@
</exclusions>
</dependency>
<dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-aws</artifactId>
+ <version>${hadoop.version}</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.orc</groupId>
<artifactId>orc-core</artifactId>
<version>1.5.2</version>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org