Repository: orc Updated Branches: refs/heads/master 4c0724312 -> c9de6f4e4
ORC-276: [C++] Create a simple tool to import CSV files Fixes #199 Signed-off-by: Deepak Majeti <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/orc/repo Commit: http://git-wip-us.apache.org/repos/asf/orc/commit/c9de6f4e Tree: http://git-wip-us.apache.org/repos/asf/orc/tree/c9de6f4e Diff: http://git-wip-us.apache.org/repos/asf/orc/diff/c9de6f4e Branch: refs/heads/master Commit: c9de6f4e4364fd60a2c0f786711ca2bc96929e01 Parents: 4c07243 Author: Gang Wu <[email protected]> Authored: Wed Dec 6 14:55:33 2017 -0800 Committer: Deepak Majeti <[email protected]> Committed: Thu Jan 11 16:04:31 2018 +0530 ---------------------------------------------------------------------- examples/TestCSVFileImport.test10rows.csv | 10 + tools/src/CMakeLists.txt | 8 + tools/src/CSVFileImport.cc | 474 +++++++++++++++++++++++++ tools/test/CMakeLists.txt | 1 + tools/test/TestCSVFileImport.cc | 55 +++ 5 files changed, 548 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/orc/blob/c9de6f4e/examples/TestCSVFileImport.test10rows.csv ---------------------------------------------------------------------- diff --git a/examples/TestCSVFileImport.test10rows.csv b/examples/TestCSVFileImport.test10rows.csv new file mode 100644 index 0000000..5404dac --- /dev/null +++ b/examples/TestCSVFileImport.test10rows.csv @@ -0,0 +1,10 @@ +0,a,0.0 +1,b,1.1 +2,c,2.2 +3,d, +4,,4.4 +,f,5.5 +,, +7,h,7.7 +8,i,8.8 +9,j,9.9 \ No newline at end of file http://git-wip-us.apache.org/repos/asf/orc/blob/c9de6f4e/tools/src/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/tools/src/CMakeLists.txt b/tools/src/CMakeLists.txt index 435a3e3..667fab4 100644 --- a/tools/src/CMakeLists.txt +++ b/tools/src/CMakeLists.txt @@ -76,6 +76,14 @@ target_link_libraries (timezone-dump orc ) +add_executable (csv-import + CSVFileImport.cc + ) + +target_link_libraries (csv-import + orc + ) + install(TARGETS orc-contents orc-metadata http://git-wip-us.apache.org/repos/asf/orc/blob/c9de6f4e/tools/src/CSVFileImport.cc ---------------------------------------------------------------------- diff --git a/tools/src/CSVFileImport.cc b/tools/src/CSVFileImport.cc new file mode 100644 index 0000000..f2eadfd --- /dev/null +++ b/tools/src/CSVFileImport.cc @@ -0,0 +1,474 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "orc/Exceptions.hh" +#include "orc/OrcFile.hh" + +#include <algorithm> +#include <fstream> +#include <iostream> +#include <memory> +#include <getopt.h> +#include <string> +#include <sys/time.h> +#include <time.h> + +static char gDelimiter = ','; + +// extract one column raw text from one line +std::string extractColumn(std::string s, uint64_t colIndex) { + uint64_t col = 0; + size_t start = 0; + size_t end = s.find(gDelimiter); + while (col < colIndex && end != std::string::npos) { + start = end + 1; + end = s.find(gDelimiter, start); + ++col; + } + return col == colIndex ? s.substr(start, end - start) : ""; +} + +static const char* GetDate(void) { + static char buf[200]; + time_t t = time(NULL); + struct tm* p = localtime(&t); + strftime(buf, sizeof(buf), "[%Y-%m-%d %H:%M:%S]", p); + return buf; +} + +void fillLongValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex) { + orc::LongVectorBatch* longBatch = + dynamic_cast<orc::LongVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + longBatch->data[i] = atoll(col.c_str()); + } + } + longBatch->hasNulls = hasNull; + longBatch->numElements = numValues; +} + +void fillStringValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex, + orc::DataBuffer<char>& buffer, + uint64_t& offset) { + orc::StringVectorBatch* stringBatch = + dynamic_cast<orc::StringVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + if (buffer.size() - offset < col.size()) { + buffer.reserve(buffer.size() * 2); + } + memcpy(buffer.data() + offset, + col.c_str(), + col.size()); + stringBatch->data[i] = buffer.data() + offset; + stringBatch->length[i] = static_cast<int64_t>(col.size()); + offset += col.size(); + } + } + stringBatch->hasNulls = hasNull; + stringBatch->numElements = numValues; +} + +void fillDoubleValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex) { + orc::DoubleVectorBatch* dblBatch = + dynamic_cast<orc::DoubleVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + dblBatch->data[i] = atof(col.c_str()); + } + } + dblBatch->hasNulls = hasNull; + dblBatch->numElements = numValues; +} + +// parse fixed point decimal numbers +void fillDecimalValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex, + size_t scale, + size_t precision) { + + + orc::Decimal128VectorBatch* d128Batch = ORC_NULLPTR; + orc::Decimal64VectorBatch* d64Batch = ORC_NULLPTR; + if (precision <= 18) { + d64Batch = dynamic_cast<orc::Decimal64VectorBatch*>(batch); + d64Batch->scale = static_cast<int32_t>(scale); + } else { + d128Batch = dynamic_cast<orc::Decimal128VectorBatch*>(batch); + d128Batch->scale = static_cast<int32_t>(scale); + } + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + size_t ptPos = col.find('.'); + size_t curScale = 0; + std::string num = col; + if (ptPos != std::string::npos) { + curScale = col.length() - ptPos - 1; + num = col.substr(0, ptPos) + col.substr(ptPos + 1); + } + orc::Int128 decimal(num); + while (curScale != scale) { + curScale++; + decimal *= 10; + } + if (precision <= 18) { + d64Batch->values[i] = decimal.toLong(); + } else { + d128Batch->values[i] = decimal; + } + } + } + batch->hasNulls = hasNull; + batch->numElements = numValues; +} + +void fillBoolValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex) { + orc::LongVectorBatch* boolBatch = + dynamic_cast<orc::LongVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + std::transform(col.begin(), col.end(), col.begin(), ::tolower); + if (col == "true" || col == "t") { + boolBatch->data[i] = true; + } else { + boolBatch->data[i] = false; + } + } + } + boolBatch->hasNulls = hasNull; + boolBatch->numElements = numValues; +} + +// parse date string from format YYYY-mm-dd +void fillDateValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex) { + orc::LongVectorBatch* longBatch = + dynamic_cast<orc::LongVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + struct tm tm; + memset(&tm, 0, sizeof(struct tm)); + strptime(col.c_str(), "%Y-%m-%d", &tm); + time_t t = mktime(&tm); + time_t t1970 = 0; + double seconds = difftime(t, t1970); + int64_t days = static_cast<int64_t>(seconds / (60*60*24)); + longBatch->data[i] = days; + } + } + longBatch->hasNulls = hasNull; + longBatch->numElements = numValues; +} + +// parse timestamp values in seconds +void fillTimestampValues(const std::vector<std::string>& data, + orc::ColumnVectorBatch* batch, + uint64_t numValues, + uint64_t colIndex) { + orc::TimestampVectorBatch* tsBatch = + dynamic_cast<orc::TimestampVectorBatch*>(batch); + bool hasNull = false; + for (uint64_t i = 0; i < numValues; ++i) { + std::string col = extractColumn(data[i], colIndex); + if (col.empty()) { + batch->notNull[i] = 0; + hasNull = true; + } else { + batch->notNull[i] = 1; + tsBatch->data[i] = atoll(col.c_str()); + tsBatch->nanoseconds[i] = 0; + } + } + tsBatch->hasNulls = hasNull; + tsBatch->numElements = numValues; +} + +void usage() { + std::cout << "Usage: csv-import [-h] [--help]\n" + << " [-d <character>] [--delimiter=<character>]\n" + << " [-s <size>] [--stripe=<size>]\n" + << " [-c <size>] [--block=<size>]\n" + << " [-b <size>] [--batch=<size>]\n" + << " <schema> <input> <output>\n" + << "Import CSV file into an Orc file using the specified schema.\n" + << "Compound types are not yet supported.\n"; +} + +int main(int argc, char* argv[]) { + std::string input; + std::string output; + std::string schema; + uint64_t stripeSize = (128 << 20); // 128M + uint64_t blockSize = 64 << 10; // 64K + uint64_t batchSize = 1024; + orc::CompressionKind compression = orc::CompressionKind_ZLIB; + + static struct option longOptions[] = { + {"help", no_argument, ORC_NULLPTR, 'h'}, + {"delimiter", required_argument, ORC_NULLPTR, 'd'}, + {"stripe", required_argument, ORC_NULLPTR, 'p'}, + {"block", required_argument, ORC_NULLPTR, 'c'}, + {"batch", required_argument, ORC_NULLPTR, 'b'}, + {ORC_NULLPTR, 0, ORC_NULLPTR, 0} + }; + bool helpFlag = false; + int opt; + char *tail; + do { + opt = getopt_long(argc, argv, "i:o:s:b:c:p:h", longOptions, ORC_NULLPTR); + switch (opt) { + case '?': + case 'h': + helpFlag = true; + opt = -1; + break; + case 'd': + gDelimiter = optarg[0]; + break; + case 'p': + stripeSize = strtoul(optarg, &tail, 10); + if (*tail != '\0') { + fprintf(stderr, "The --stripe parameter requires an integer option.\n"); + return 1; + } + break; + case 'c': + blockSize = strtoul(optarg, &tail, 10); + if (*tail != '\0') { + fprintf(stderr, "The --block parameter requires an integer option.\n"); + return 1; + } + break; + case 'b': + batchSize = strtoul(optarg, &tail, 10); + if (*tail != '\0') { + fprintf(stderr, "The --batch parameter requires an integer option.\n"); + return 1; + } + break; + } + } while (opt != -1); + + argc -= optind; + argv += optind; + + if (argc != 3 || helpFlag) { + usage(); + return 1; + } + + schema = argv[0]; + input = argv[1]; + output = argv[2]; + + std::cout << GetDate() << " Start importing Orc file..." << std::endl; + ORC_UNIQUE_PTR<orc::Type> fileType = orc::Type::buildTypeFromString(schema); + + double totalElapsedTime = 0.0; + clock_t totalCPUTime = 0; + + orc::DataBuffer<char> buffer(*orc::getDefaultPool(), 4 * 1024 * 1024); + + orc::WriterOptions options; + options.setStripeSize(stripeSize); + options.setCompressionBlockSize(blockSize); + options.setCompression(compression); + + ORC_UNIQUE_PTR<orc::OutputStream> outStream = orc::writeLocalFile(output); + ORC_UNIQUE_PTR<orc::Writer> writer = + orc::createWriter(*fileType, outStream.get(), options); + ORC_UNIQUE_PTR<orc::ColumnVectorBatch> rowBatch = + writer->createRowBatch(batchSize); + + bool eof = false; + std::string line; + std::vector<std::string> data; // buffer that holds a batch of rows in raw text + std::ifstream finput(input.c_str()); + while (!eof) { + uint64_t numValues = 0; // num of lines read in a batch + uint64_t bufferOffset = 0; // current offset in the string buffer + + data.clear(); + memset(rowBatch->notNull.data(), 1, batchSize); + + // read a batch of lines from the input file + for (uint64_t i = 0; i < batchSize; ++i) { + if (!std::getline(finput, line)) { + eof = true; + break; + } + data.push_back(line); + ++numValues; + } + + if (numValues != 0) { + orc::StructVectorBatch* structBatch = + dynamic_cast<orc::StructVectorBatch*>(rowBatch.get()); + structBatch->numElements = numValues; + + for (uint64_t i = 0; i < structBatch->fields.size(); ++i) { + const orc::Type* subType = fileType->getSubtype(i); + switch (subType->getKind()) { + case orc::BYTE: + case orc::INT: + case orc::SHORT: + case orc::LONG: + fillLongValues(data, + structBatch->fields[i], + numValues, + i); + break; + case orc::STRING: + case orc::CHAR: + case orc::VARCHAR: + case orc::BINARY: + fillStringValues(data, + structBatch->fields[i], + numValues, + i, + buffer, + bufferOffset); + break; + case orc::FLOAT: + case orc::DOUBLE: + fillDoubleValues(data, + structBatch->fields[i], + numValues, + i); + break; + case orc::DECIMAL: + fillDecimalValues(data, + structBatch->fields[i], + numValues, + i, + subType->getScale(), + subType->getPrecision()); + break; + case orc::BOOLEAN: + fillBoolValues(data, + structBatch->fields[i], + numValues, + i); + break; + case orc::DATE: + fillDateValues(data, + structBatch->fields[i], + numValues, + i); + break; + case orc::TIMESTAMP: + fillTimestampValues(data, + structBatch->fields[i], + numValues, + i); + break; + case orc::STRUCT: + case orc::LIST: + case orc::MAP: + case orc::UNION: + throw std::runtime_error(subType->toString() + " is not supported yet."); + } + } + + struct timeval t_start, t_end; + gettimeofday(&t_start, NULL); + clock_t c_start = clock(); + + writer->add(*rowBatch); + + totalCPUTime += (clock() - c_start); + gettimeofday(&t_end, NULL); + totalElapsedTime += + (static_cast<double>(t_end.tv_sec - t_start.tv_sec) * 1000000.0 + + static_cast<double>(t_end.tv_usec - t_start.tv_usec)) / 1000000.0; + } + } + + struct timeval t_start, t_end; + gettimeofday(&t_start, NULL); + clock_t c_start = clock(); + + writer->close(); + + totalCPUTime += (clock() - c_start); + gettimeofday(&t_end, NULL); + totalElapsedTime += + (static_cast<double>(t_end.tv_sec - t_start.tv_sec) * 1000000.0 + + static_cast<double>(t_end.tv_usec - t_start.tv_usec)) / 1000000.0; + + std::cout << GetDate() << " Finish importing Orc file." << std::endl; + std::cout << GetDate() << " Total writer elasped time: " + << totalElapsedTime << "s." << std::endl; + std::cout << GetDate() << " Total writer CPU time: " + << static_cast<double>(totalCPUTime) / CLOCKS_PER_SEC + << "s." << std::endl; + return 0; +} http://git-wip-us.apache.org/repos/asf/orc/blob/c9de6f4e/tools/test/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/tools/test/CMakeLists.txt b/tools/test/CMakeLists.txt index cd038be..6664b83 100644 --- a/tools/test/CMakeLists.txt +++ b/tools/test/CMakeLists.txt @@ -25,6 +25,7 @@ set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${CXX11_FLAGS} ${WARN_FLAGS}") add_executable (tool-test gzip.cc + TestCSVFileImport.cc TestFileMetadata.cc TestFileScan.cc TestMatch.cc http://git-wip-us.apache.org/repos/asf/orc/blob/c9de6f4e/tools/test/TestCSVFileImport.cc ---------------------------------------------------------------------- diff --git a/tools/test/TestCSVFileImport.cc b/tools/test/TestCSVFileImport.cc new file mode 100644 index 0000000..2dd1b4b --- /dev/null +++ b/tools/test/TestCSVFileImport.cc @@ -0,0 +1,55 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "orc/OrcFile.hh" + +#include "Adaptor.hh" +#include "ToolTest.hh" + +#include "wrap/gmock.h" +#include "wrap/gtest-wrapper.h" + +TEST (TestCSVFileImport, test10rows) { + // create an ORC file from importing the CSV file + const std::string pgm1 = findProgram("tools/src/csv-import"); + const std::string csvFile = findExample("TestCSVFileImport.test10rows.csv"); + const std::string orcFile = "/tmp/test_csv_import_test_10_rows.orc"; + const std::string schema = "struct<a:bigint,b:string,c:double>"; + std::string output; + std::string error; + + EXPECT_EQ(0, runProgram({pgm1, schema, csvFile, orcFile}, output, error)); + EXPECT_EQ("", error); + + // verify the ORC file content + const std::string pgm2 = findProgram("tools/src/orc-contents"); + const std::string expected = + "{\"a\": 0, \"b\": \"a\", \"c\": 0}\n" + "{\"a\": 1, \"b\": \"b\", \"c\": 1.1}\n" + "{\"a\": 2, \"b\": \"c\", \"c\": 2.2}\n" + "{\"a\": 3, \"b\": \"d\", \"c\": null}\n" + "{\"a\": 4, \"b\": null, \"c\": 4.4}\n" + "{\"a\": null, \"b\": \"f\", \"c\": 5.5}\n" + "{\"a\": null, \"b\": null, \"c\": null}\n" + "{\"a\": 7, \"b\": \"h\", \"c\": 7.7}\n" + "{\"a\": 8, \"b\": \"i\", \"c\": 8.8}\n" + "{\"a\": 9, \"b\": \"j\", \"c\": 9.9}\n"; + EXPECT_EQ(0, runProgram({pgm2, orcFile}, output, error)); + EXPECT_EQ(expected, output); + EXPECT_EQ("", error); +}
