| /** |
| * 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. |
| */ |
| |
| package org.apache.falcon.regression.hcat; |
| |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.falcon.regression.Entities.FeedMerlin; |
| import org.apache.falcon.regression.core.bundle.Bundle; |
| import org.apache.falcon.entity.v0.cluster.Interfacetype; |
| import org.apache.falcon.entity.v0.EntityType; |
| import org.apache.falcon.entity.v0.Frequency; |
| import org.apache.falcon.entity.v0.process.EngineType; |
| import org.apache.falcon.regression.core.helpers.ColoHelper; |
| import org.apache.falcon.regression.core.util.AssertUtil; |
| import org.apache.falcon.regression.core.util.BundleUtil; |
| import org.apache.falcon.regression.core.util.HCatUtil; |
| import org.apache.falcon.regression.core.util.HadoopUtil; |
| import org.apache.falcon.regression.core.util.InstanceUtil; |
| import org.apache.falcon.regression.core.util.OSUtil; |
| import org.apache.falcon.regression.core.util.TimeUtil; |
| import org.apache.falcon.regression.core.util.Util; |
| import org.apache.falcon.regression.testHelper.BaseTestClass; |
| import org.apache.hadoop.fs.ContentSummary; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| |
| import org.apache.hive.hcatalog.data.schema.HCatFieldSchema; |
| import org.apache.hive.hcatalog.api.HCatAddPartitionDesc; |
| import org.apache.hive.hcatalog.api.HCatClient; |
| import org.apache.hive.hcatalog.api.HCatCreateTableDesc; |
| import org.apache.hive.hcatalog.common.HCatException; |
| import org.apache.log4j.Logger; |
| import org.apache.oozie.client.CoordinatorAction; |
| import org.apache.oozie.client.OozieClient; |
| import org.joda.time.DateTime; |
| import org.joda.time.format.DateTimeFormat; |
| import org.joda.time.format.DateTimeFormatter; |
| import org.testng.Assert; |
| import org.testng.annotations.AfterMethod; |
| import org.testng.annotations.BeforeMethod; |
| import org.testng.annotations.DataProvider; |
| import org.testng.annotations.Test; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| @Test(groups = "embedded") |
| public class HCatProcessTest extends BaseTestClass { |
| private static final Logger logger = Logger.getLogger(HCatProcessTest.class); |
| ColoHelper cluster = servers.get(0); |
| FileSystem clusterFS = serverFS.get(0); |
| OozieClient clusterOC = serverOC.get(0); |
| HCatClient clusterHC; |
| |
| final String testDir = "/HCatProcessTest"; |
| final String baseTestHDFSDir = baseHDFSDir + testDir; |
| String hiveScriptDir = baseTestHDFSDir + "/hive"; |
| String hiveScriptFile = hiveScriptDir + "/script.hql"; |
| String aggregateWorkflowDir = baseTestHDFSDir + "/aggregator"; |
| String hiveScriptFileNonHCatInput = hiveScriptDir + "/script_non_hcat_input.hql"; |
| String hiveScriptFileNonHCatOutput = hiveScriptDir + "/script_non_hcat_output.hql"; |
| String hiveScriptTwoHCatInputOneHCatOutput = |
| hiveScriptDir + "/script_two_hcat_input_one_hcat_output.hql"; |
| String hiveScriptOneHCatInputTwoHCatOutput = |
| hiveScriptDir + "/script_one_hcat_input_two_hcat_output.hql"; |
| String hiveScriptTwoHCatInputTwoHCatOutput = |
| hiveScriptDir + "/script_two_hcat_input_two_hcat_output.hql"; |
| final String inputHDFSDir = baseTestHDFSDir + "/input"; |
| final String inputHDFSDir2 = baseTestHDFSDir + "/input2"; |
| final String outputHDFSDir = baseTestHDFSDir + "/output"; |
| final String outputHDFSDir2 = baseTestHDFSDir + "/output2"; |
| |
| final String dbName = "default"; |
| final String inputTableName = "hcatprocesstest_input_table"; |
| final String inputTableName2 = "hcatprocesstest_input_table2"; |
| final String outputTableName = "hcatprocesstest_output_table"; |
| final String outputTableName2 = "hcatprocesstest_output_table2"; |
| public static final String col1Name = "id"; |
| public static final String col2Name = "value"; |
| public static final String partitionColumn = "dt"; |
| |
| private static final String hcatDir = OSUtil.getPath("src", "test", "resources", "hcat"); |
| private static final String localHCatData = OSUtil.getPath(hcatDir, "data"); |
| private static final String hiveScript = OSUtil.getPath(hcatDir, "hivescript"); |
| |
| @BeforeMethod(alwaysRun = true) |
| public void setUp() throws Exception { |
| clusterHC = cluster.getClusterHelper().getHCatClient(); |
| bundles[0] = BundleUtil.readHCatBundle(baseAppHDFSDir, this.getClass().getSimpleName()); |
| bundles[0] = new Bundle(bundles[0], cluster); |
| bundles[0].generateUniqueBundle(); |
| bundles[0].setProcessWorkflow(hiveScriptFile, EngineType.HIVE); |
| bundles[0].setClusterInterface(Interfacetype.REGISTRY, |
| cluster.getClusterHelper().getHCatEndpoint()); |
| |
| HadoopUtil.deleteDirIfExists(baseTestHDFSDir, clusterFS); |
| HadoopUtil.uploadDir(clusterFS, hiveScriptDir, hiveScript); |
| HadoopUtil.uploadDir(clusterFS, aggregateWorkflowDir, OSUtil.RESOURCES_OOZIE); |
| HadoopUtil.recreateDir(clusterFS, outputHDFSDir); |
| HadoopUtil.recreateDir(clusterFS, outputHDFSDir2); |
| clusterHC.dropTable(dbName, inputTableName, true); |
| clusterHC.dropTable(dbName, inputTableName2, true); |
| clusterHC.dropTable(dbName, outputTableName, true); |
| clusterHC.dropTable(dbName, outputTableName2, true); |
| } |
| |
| @DataProvider |
| public String[][] generateSeparators() { |
| //disabling till FALCON-372 is fixed |
| //return new String[][] {{"-"}, {"/"}}; |
| return new String[][]{{"-"},}; |
| } |
| |
| @Test(dataProvider = "generateSeparators") |
| public void OneHCatInputOneHCatOutput(String separator) throws Exception { |
| /* upload data and create partition */ |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir) |
| .build()); |
| |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, inputTableName); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String inputTableUri = |
| "catalog:" + dbName + ":" + inputTableName + tableUriPartitionFragment; |
| bundles[0].setInputFeedTableUri(inputTableUri); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| String outputTableUri = |
| "catalog:" + dbName + ":" + outputTableName + tableUriPartitionFragment; |
| bundles[0].setOutputFeedTableUri(outputTableUri); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| AssertUtil.checkContentSize(inputHDFSDir + "/" + dataDates.get(0), |
| outputHDFSDir + "/dt=" + dataDates.get(0), clusterFS); |
| } |
| |
| @Test(dataProvider = "generateSeparators") |
| public void TwoHCatInputOneHCatOutput(String separator) throws Exception { |
| /* upload data and create partition */ |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| final List<String> dataset2 = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir2, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName2, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir2) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir) |
| .build()); |
| |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, inputTableName); |
| addPartitionsToTable(dataDates, dataset2, "dt", dbName, inputTableName2); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String inputTableUri = |
| "catalog:" + dbName + ":" + inputTableName + tableUriPartitionFragment; |
| String inputTableUri2 = |
| "catalog:" + dbName + ":" + inputTableName2 + tableUriPartitionFragment; |
| bundles[0].setInputFeedTableUri(inputTableUri); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| final String inputFeed1 = bundles[0].getInputFeedFromBundle(); |
| final String inputFeed2Name = "second-" + Util.readEntityName(inputFeed1); |
| |
| FeedMerlin feedObj = new FeedMerlin(inputFeed1); |
| feedObj.setName(inputFeed2Name); |
| feedObj.getTable().setUri(inputTableUri2); |
| |
| String inputFeed2 = feedObj.toString(); |
| bundles[0].addInputFeedToBundle("inputData2", inputFeed2, 0); |
| |
| String outputTableUri = |
| "catalog:" + dbName + ":" + outputTableName + tableUriPartitionFragment; |
| bundles[0].setOutputFeedTableUri(outputTableUri); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| bundles[0].setProcessWorkflow(hiveScriptTwoHCatInputOneHCatOutput, EngineType.HIVE); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| final ContentSummary inputContentSummary = |
| clusterFS.getContentSummary(new Path(inputHDFSDir + "/" + dataDates.get(0))); |
| final ContentSummary inputContentSummary2 = |
| clusterFS.getContentSummary(new Path(inputHDFSDir2 + "/" + dataDates.get(0))); |
| final ContentSummary outputContentSummary = |
| clusterFS.getContentSummary(new Path(outputHDFSDir + "/dt=" + dataDates.get(0))); |
| logger.info("inputContentSummary = " + inputContentSummary.toString(false)); |
| logger.info("inputContentSummary2 = " + inputContentSummary2.toString(false)); |
| logger.info("outputContentSummary = " + outputContentSummary.toString(false)); |
| Assert.assertEquals(inputContentSummary.getLength() + inputContentSummary2.getLength(), |
| outputContentSummary.getLength(), |
| "Unexpected size of the output."); |
| } |
| |
| @Test(dataProvider = "generateSeparators") |
| public void OneHCatInputTwoHCatOutput(String separator) throws Exception { |
| /* upload data and create partition */ |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName2, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir2) |
| .build()); |
| |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, inputTableName); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String inputTableUri = |
| "catalog:" + dbName + ":" + inputTableName + tableUriPartitionFragment; |
| bundles[0].setInputFeedTableUri(inputTableUri); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| String outputTableUri = |
| "catalog:" + dbName + ":" + outputTableName + tableUriPartitionFragment; |
| String outputTableUri2 = |
| "catalog:" + dbName + ":" + outputTableName2 + tableUriPartitionFragment; |
| bundles[0].setOutputFeedTableUri(outputTableUri); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| final String outputFeed1 = bundles[0].getInputFeedFromBundle(); |
| final String outputFeed2Name = "second-" + Util.readEntityName(outputFeed1); |
| FeedMerlin feedObj = new FeedMerlin(outputFeed1); |
| feedObj.setName(outputFeed2Name); |
| feedObj.getTable().setUri(outputTableUri2); |
| bundles[0].addOutputFeedToBundle("outputData2", feedObj.toString(), 0); |
| |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| bundles[0].setProcessWorkflow(hiveScriptOneHCatInputTwoHCatOutput, EngineType.HIVE); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| AssertUtil.checkContentSize(inputHDFSDir + "/" + dataDates.get(0), |
| outputHDFSDir + "/dt=" + dataDates.get(0), clusterFS); |
| AssertUtil.checkContentSize(inputHDFSDir + "/" + dataDates.get(0), |
| outputHDFSDir2 + "/dt=" + dataDates.get(0), clusterFS); |
| } |
| |
| |
| @Test(dataProvider = "generateSeparators") |
| public void TwoHCatInputTwoHCatOutput(String separator) throws Exception { |
| /* upload data and create partition */ |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| final List<String> dataset2 = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir2, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName2, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir2) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir) |
| .build()); |
| |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName2, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir2) |
| .build()); |
| |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, inputTableName); |
| addPartitionsToTable(dataDates, dataset2, "dt", dbName, inputTableName2); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String inputTableUri = |
| "catalog:" + dbName + ":" + inputTableName + tableUriPartitionFragment; |
| String inputTableUri2 = |
| "catalog:" + dbName + ":" + inputTableName2 + tableUriPartitionFragment; |
| bundles[0].setInputFeedTableUri(inputTableUri); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| final String inputFeed1 = bundles[0].getInputFeedFromBundle(); |
| final String inputFeed2Name = "second-" + Util.readEntityName(inputFeed1); |
| FeedMerlin feedObj = new FeedMerlin(inputFeed1); |
| feedObj.setName(inputFeed2Name); |
| feedObj.getTable().setUri(inputTableUri2); |
| String inputFeed2 = feedObj.toString(); |
| bundles[0].addInputFeedToBundle("inputData2", inputFeed2, 0); |
| |
| String outputTableUri = |
| "catalog:" + dbName + ":" + outputTableName + tableUriPartitionFragment; |
| String outputTableUri2 = |
| "catalog:" + dbName + ":" + outputTableName2 + tableUriPartitionFragment; |
| bundles[0].setOutputFeedTableUri(outputTableUri); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| final String outputFeed1 = bundles[0].getOutputFeedFromBundle(); |
| final String outputFeed2Name = "second-" + Util.readEntityName(outputFeed1); |
| FeedMerlin feedObj2 = new FeedMerlin(outputFeed1); |
| feedObj2.setName(outputFeed2Name); |
| feedObj2.getTable().setUri(outputTableUri2); |
| String outputFeed2 = feedObj2.toString(); |
| bundles[0].addOutputFeedToBundle("outputData2", outputFeed2, 0); |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| bundles[0].setProcessWorkflow(hiveScriptTwoHCatInputTwoHCatOutput, EngineType.HIVE); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| final ContentSummary inputContentSummary = |
| clusterFS.getContentSummary(new Path(inputHDFSDir + "/" + dataDates.get(0))); |
| final ContentSummary inputContentSummary2 = |
| clusterFS.getContentSummary(new Path(inputHDFSDir2 + "/" + dataDates.get(0))); |
| final ContentSummary outputContentSummary = |
| clusterFS.getContentSummary(new Path(outputHDFSDir + "/dt=" + dataDates.get(0))); |
| final ContentSummary outputContentSummary2 = |
| clusterFS.getContentSummary(new Path(outputHDFSDir2 + "/dt=" + dataDates.get(0))); |
| logger.info("inputContentSummary = " + inputContentSummary.toString(false)); |
| logger.info("inputContentSummary2 = " + inputContentSummary2.toString(false)); |
| logger.info("outputContentSummary = " + outputContentSummary.toString(false)); |
| logger.info("outputContentSummary2 = " + outputContentSummary2.toString(false)); |
| Assert.assertEquals(inputContentSummary.getLength() + inputContentSummary2.getLength(), |
| outputContentSummary.getLength(), |
| "Unexpected size of the output."); |
| Assert.assertEquals(inputContentSummary.getLength() + inputContentSummary2.getLength(), |
| outputContentSummary2.getLength(), |
| "Unexpected size of the output."); |
| } |
| |
| |
| @Test(dataProvider = "generateSeparators") |
| public void OneHCatInputOneNonHCatOutput(String separator) throws Exception { |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| /* upload data and create partition */ |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil |
| .flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, inputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(inputHDFSDir) |
| .build()); |
| |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, inputTableName); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String inputTableUri = |
| "catalog:" + dbName + ":" + inputTableName + tableUriPartitionFragment; |
| bundles[0].setInputFeedTableUri(inputTableUri); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| |
| // |
| String nonHCatFeed = BundleUtil.readELBundle( |
| baseAppHDFSDir, this.getClass().getSimpleName()).getOutputFeedFromBundle(); |
| final String outputFeedName = bundles[0].getOutputFeedNameFromBundle(); |
| nonHCatFeed = Util.setFeedName(nonHCatFeed, outputFeedName); |
| final List<String> clusterNames = bundles[0].getClusterNames(); |
| Assert.assertEquals(clusterNames.size(), 1, "Expected only one cluster in the bundle."); |
| nonHCatFeed = Util.setClusterNameInFeed(nonHCatFeed, clusterNames.get(0), 0); |
| bundles[0].writeFeedElement(nonHCatFeed, outputFeedName); |
| bundles[0].setOutputFeedLocationData(outputHDFSDir + "/" + |
| StringUtils.join(new String[]{"${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator)); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| |
| bundles[0].setProcessWorkflow(hiveScriptFileNonHCatOutput, EngineType.HIVE); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| AssertUtil.checkContentSize(inputHDFSDir + "/" + dataDates.get(0), |
| outputHDFSDir + "/" + dataDates.get(0), clusterFS); |
| } |
| |
| @Test(dataProvider = "generateSeparators") |
| public void OneNonCatInputOneHCatOutput(String separator) throws Exception { |
| /* upload data and create partition */ |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2010-01-02T04:00Z"; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| List<String> dataDates = getDatesList(startDate, endDate, datePattern, 60); |
| |
| final List<String> dataset = HadoopUtil. |
| flattenAndPutDataInFolder(clusterFS, localHCatData, inputHDFSDir, dataDates); |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| clusterHC.createTable(HCatCreateTableDesc |
| .create(dbName, outputTableName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(outputHDFSDir) |
| .build()); |
| |
| String nonHCatFeed = BundleUtil.readELBundle( |
| baseAppHDFSDir, this.getClass().getSimpleName()).getInputFeedFromBundle(); |
| final String inputFeedName = bundles[0].getInputFeedNameFromBundle(); |
| nonHCatFeed = Util.setFeedName(nonHCatFeed, inputFeedName); |
| final List<String> clusterNames = bundles[0].getClusterNames(); |
| Assert.assertEquals(clusterNames.size(), 1, "Expected only one cluster in the bundle."); |
| nonHCatFeed = Util.setClusterNameInFeed(nonHCatFeed, clusterNames.get(0), 0); |
| bundles[0].writeFeedElement(nonHCatFeed, inputFeedName); |
| bundles[0].setInputFeedDataPath(inputHDFSDir + "/" + |
| StringUtils.join(new String[]{"${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator)); |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| |
| final String tableUriPartitionFragment = StringUtils.join( |
| new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String outputTableUri = |
| "catalog:" + dbName + ":" + outputTableName + tableUriPartitionFragment; |
| bundles[0].setOutputFeedTableUri(outputTableUri); |
| bundles[0].setOutputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setOutputFeedValidity(startDate, endDate); |
| |
| bundles[0].setProcessWorkflow(hiveScriptFileNonHCatInput, EngineType.HIVE); |
| bundles[0].setProcessValidity(startDate, endDate); |
| bundles[0].setProcessPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setProcessInputStartEnd("now(0,0)", "now(0,0)"); |
| bundles[0].submitFeedsScheduleProcess(); |
| |
| InstanceUtil.waitTillInstanceReachState( |
| clusterOC, bundles[0].getProcessName(), 1, CoordinatorAction.Status.SUCCEEDED, |
| EntityType.PROCESS); |
| |
| AssertUtil.checkContentSize(inputHDFSDir + "/" + dataDates.get(0), |
| outputHDFSDir + "/dt=" + dataDates.get(0), clusterFS); |
| } |
| |
| private void addPartitionsToTable(List<String> partitions, List<String> partitionLocations, |
| String partitionCol, |
| String dbName, String tableName) throws HCatException { |
| Assert.assertEquals(partitions.size(), partitionLocations.size(), |
| "Number of locations is not same as number of partitions."); |
| final List<HCatAddPartitionDesc> partitionDesc = new ArrayList<HCatAddPartitionDesc>(); |
| for (int i = 0; i < partitions.size(); ++i) { |
| final String partition = partitions.get(i); |
| final Map<String, String> onePartition = new HashMap<String, String>(); |
| onePartition.put(partitionCol, partition); |
| final String partitionLoc = partitionLocations.get(i); |
| partitionDesc.add( |
| HCatAddPartitionDesc.create(dbName, tableName, partitionLoc, onePartition).build()); |
| } |
| clusterHC.addPartitions(partitionDesc); |
| } |
| |
| public static List<String> getDatesList(String startDate, String endDate, String datePattern, |
| int skipMinutes) { |
| DateTime startDateJoda = new DateTime(TimeUtil.oozieDateToDate(startDate)); |
| DateTime endDateJoda = new DateTime(TimeUtil.oozieDateToDate(endDate)); |
| DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); |
| logger.info("generating data between " + formatter.print(startDateJoda) + " and " + |
| formatter.print(endDateJoda)); |
| List<String> dates = new ArrayList<String>(); |
| dates.add(formatter.print(startDateJoda)); |
| while (!startDateJoda.isAfter(endDateJoda)) { |
| startDateJoda = startDateJoda.plusMinutes(skipMinutes); |
| dates.add(formatter.print(startDateJoda)); |
| } |
| return dates; |
| } |
| |
| @AfterMethod(alwaysRun = true) |
| public void tearDown() { |
| removeBundles(); |
| } |
| } |
