hive/src/main/java/org/apache/apex/malhar/hive/HiveOutputModule.java

/**
 * 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.apex.malhar.hive;

import java.util.ArrayList;
import java.util.Arrays;

import javax.annotation.Nonnull;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;

import org.apache.apex.malhar.hive.FSPojoToHiveOperator.FIELD_TYPE;
import org.apache.commons.beanutils.ConvertUtils;
import org.apache.commons.beanutils.converters.AbstractConverter;
import org.apache.commons.beanutils.converters.ArrayConverter;
import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.DAG;
import com.datatorrent.api.Module;

/**
 * HiveOutputModule provides abstraction for the operators needed for writing
 * tuples to hive. This module will be expanded to FSPojoToHiveOperator and
 * HiveOperator in physical plan.
 *
 * @since 3.5.0
 */
@org.apache.hadoop.classification.InterfaceStability.Evolving
public class HiveOutputModule implements Module
{

  /**
   * The path of the directory to where files are written.
   */
  @NotNull
  private String filePath;

  /**
   * Names of the columns in hive table (excluding partitioning columns).
   */
  private String[] hiveColumns;

  /**
   * Data types of the columns in hive table (excluding partitioning columns).
   * This sequence should match to the fields in hiveColumnDataTypes
   */
  private FIELD_TYPE[] hiveColumnDataTypes;

  /**
   * Expressions for the hive columns (excluding partitioning columns). This
   * sequence should match to the fields in hiveColumnDataTypes
   */
  private String[] expressionsForHiveColumns;

  /**
   * Names of the columns on which hive data should be partitioned
   */
  private String[] hivePartitionColumns;

  /**
   * Data types of the columns on which hive data should be partitioned. This
   * sequence should match to the fields in hivePartitionColumns
   */
  private FIELD_TYPE[] hivePartitionColumnDataTypes;

  /**
   * Expressions for the hive partition columns. This sequence should match to
   * the fields in hivePartitionColumns
   */
  private String[] expressionsForHivePartitionColumns;

  /**
   * The maximum length in bytes of a rolling file. Default value is 128MB.
   */
  @Min(1)
  protected Long maxLength = 134217728L;

  /**
   * Connection URL for connecting to hive.
   */
  @NotNull
  private String databaseUrl;

  /**
   * Driver for connecting to hive.
   */
  @NotNull
  private String databaseDriver;

  /**
   * Username for connecting to hive
   */
  private String userName;

  /**
   * Password for connecting to hive
   */
  private String password;

  /**
   * Table name for writing data into hive
   */
  @Nonnull
  protected String tablename;

  /**
   * Input port for files metadata.
   */
  public final transient ProxyInputPort<Object> input = new ProxyInputPort<Object>();

  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {
    FSPojoToHiveOperator fsRolling = dag.addOperator("fsRolling", new FSPojoToHiveOperator());
    HiveOperator hiveOperator = dag.addOperator("HiveOperator", new HiveOperator());

    input.set(fsRolling.input);
    dag.addStream("toHive", fsRolling.outputPort, hiveOperator.input);
    fsRolling.setFilePath(filePath);
    fsRolling.setHiveColumns(new ArrayList<String>(Arrays.asList(hiveColumns)));
    fsRolling.setHiveColumnDataTypes(new ArrayList<FIELD_TYPE>(Arrays.asList(hiveColumnDataTypes)));
    fsRolling.setExpressionsForHiveColumns(new ArrayList<String>(Arrays.asList(expressionsForHiveColumns)));

    fsRolling.setHivePartitionColumns(new ArrayList<String>(Arrays.asList(hivePartitionColumns)));
    fsRolling.setHivePartitionColumnDataTypes(new ArrayList<FIELD_TYPE>(Arrays.asList(hivePartitionColumnDataTypes)));
    fsRolling.setExpressionsForHivePartitionColumns(
        new ArrayList<String>(Arrays.asList(expressionsForHivePartitionColumns)));

    fsRolling.setMaxLength(maxLength);
    fsRolling.setAlwaysWriteToTmp(true);
    fsRolling.setRotationWindows(0);

    hiveOperator.setHivePartitionColumns(new ArrayList<String>(Arrays.asList(hivePartitionColumns)));
    HiveStore hiveStore = hiveOperator.getStore();
    hiveStore.setFilepath(filePath);
    hiveStore.setDatabaseUrl(databaseUrl);
    hiveStore.setDatabaseDriver(databaseDriver);
    hiveStore.getConnectionProperties().put("user", userName);
    if (password != null) {
      hiveStore.getConnectionProperties().put("password", password);
    }
    hiveOperator.setTablename(tablename);
  }

  /**
   * The path of the directory to where files are written.
   *
   * @return file path
   */
  public String getFilePath()
  {
    return filePath;
  }

  /**
   * The path of the directory to where files are written.
   *
   * @param filePath
   *          file path
   */
  public void setFilePath(String filePath)
  {
    this.filePath = filePath;
  }

  /**
   * Names of the columns in hive table (excluding partitioning columns).
   *
   * @return Hive column names
   */
  public String[] getHiveColumns()
  {
    return hiveColumns;
  }

  /**
   * Names of the columns in hive table (excluding partitioning columns).
   *
   * @param hiveColumns
   *          Hive column names
   */
  public void setHiveColumns(String[] hiveColumns)
  {
    this.hiveColumns = hiveColumns;
  }

  /**
   * Data types of the columns in hive table (excluding partitioning columns).
   * This sequence should match to the fields in hiveColumnDataTypes
   *
   * @return Hive column data types
   */
  public FIELD_TYPE[] getHiveColumnDataTypes()
  {
    return hiveColumnDataTypes;
  }

  /**
   * Data types of the columns in hive table (excluding partitioning columns).
   * This sequence should match to the fields in hiveColumnDataTypes *
   *
   * @param hiveColumnDataTypes
   *          Hive column data types
   */
  public void setHiveColumnDataTypes(FIELD_TYPE[] hiveColumnDataTypes)
  {
    this.hiveColumnDataTypes = hiveColumnDataTypes;
  }

  /**
   * Expressions for the hive columns (excluding partitioning columns). This
   * sequence should match to the fields in hiveColumnDataTypes
   *
   * @return
   */
  public String[] getExpressionsForHiveColumns()
  {
    return expressionsForHiveColumns;
  }

  /**
   * Expressions for the hive columns (excluding partitioning columns). This
   * sequence should match to the fields in hiveColumnDataTypes
   *
   * @param expressionsForHiveColumns
   */
  public void setExpressionsForHiveColumns(String[] expressionsForHiveColumns)
  {
    this.expressionsForHiveColumns = expressionsForHiveColumns;
  }

  /**
   * Names of the columns on which hive data should be partitioned
   *
   * @return hive partition columns
   */
  public String[] getHivePartitionColumns()
  {
    return hivePartitionColumns;
  }

  /**
   * Names of the columns on which hive data should be partitioned
   *
   * @param hivePartitionColumns
   *          Hive partition columns
   */
  public void setHivePartitionColumns(String[] hivePartitionColumns)
  {
    this.hivePartitionColumns = hivePartitionColumns;
  }

  /**
   * Data types of the columns on which hive data should be partitioned. This
   * sequence should match to the fields in hivePartitionColumns
   *
   * @return Hive partition column data types
   */
  public FIELD_TYPE[] getHivePartitionColumnDataTypes()
  {
    return hivePartitionColumnDataTypes;
  }

  /**
   * Data types of the columns on which hive data should be partitioned. This
   * sequence should match to the fields in hivePartitionColumns
   *
   * @param hivePartitionColumnDataTypes
   *          Hive partition column data types
   */
  public void setHivePartitionColumnDataTypes(FIELD_TYPE[] hivePartitionColumnDataTypes)
  {
    this.hivePartitionColumnDataTypes = hivePartitionColumnDataTypes;
  }

  /**
   * Expressions for the hive partition columns. This sequence should match to
   * the fields in hivePartitionColumns
   *
   * @return Expressions for hive partition columns
   */
  public String[] getExpressionsForHivePartitionColumns()
  {
    return expressionsForHivePartitionColumns;
  }

  /**
   * Expressions for the hive partition columns. This sequence should match to
   * the fields in hivePartitionColumns
   *
   * @param expressionsForHivePartitionColumns
   *          Expressions for hive partition columns
   */
  public void setExpressionsForHivePartitionColumns(String[] expressionsForHivePartitionColumns)
  {
    this.expressionsForHivePartitionColumns = expressionsForHivePartitionColumns;
  }

  /**
   * The maximum length in bytes of a rolling file.
   *
   * @return maximum size of file
   */
  public Long getMaxLength()
  {
    return maxLength;
  }

  /**
   * The maximum length in bytes of a rolling file.
   *
   * @param maxLength
   *          maximum size of file
   */
  public void setMaxLength(Long maxLength)
  {
    this.maxLength = maxLength;
  }

  /**
   * Connection URL for connecting to hive.
   *
   * @return database url
   */
  public String getDatabaseUrl()
  {
    return databaseUrl;
  }

  /**
   * Connection URL for connecting to hive.
   *
   * @param databaseUrl
   *          database url
   */
  public void setDatabaseUrl(String databaseUrl)
  {
    this.databaseUrl = databaseUrl;
  }

  /**
   * Driver for connecting to hive.
   *
   * @return database driver
   */
  public String getDatabaseDriver()
  {
    return databaseDriver;
  }

  /**
   * Driver for connecting to hive.
   *
   * @param databaseDriver
   *          database driver
   */
  public void setDatabaseDriver(String databaseDriver)
  {
    this.databaseDriver = databaseDriver;
  }

  /**
   * Username for connecting to hive
   *
   * @return user name
   */
  public String getUserName()
  {
    return userName;
  }

  /**
   * Username for connecting to hive
   *
   * @param username
   *          user name
   */
  public void setUserName(String userName)
  {
    this.userName = userName;
  }

  /**
   * Password for connecting to hive
   *
   * @return password
   */
  public String getPassword()
  {
    return password;
  }

  /**
   * Password for connecting to hive
   *
   * @param password
   *          password
   */
  public void setPassword(String password)
  {
    this.password = password;
  }

  /**
   * Table name for writing data into hive
   *
   * @return table name
   */
  public String getTablename()
  {
    return tablename;
  }

  /**
   * Table name for writing data into hive
   *
   * @param tablename
   *          table name
   */
  public void setTablename(String tablename)
  {
    this.tablename = tablename;
  }

  static {
    //Code for enabling BeanUtils to accept comma separated string to initialize FIELD_TYPE[]
    class FieldTypeConvertor extends AbstractConverter
    {

      @Override
      protected Object convertToType(Class type, Object value) throws Throwable
      {
        if (value instanceof String) {

          return FIELD_TYPE.valueOf((String)value);
        } else {
          throw new IllegalArgumentException("FIELD_TYPE should be specified as String");
        }
      }

      @Override
      protected Class getDefaultType()
      {
        return FIELD_TYPE.class;
      }
    }

    class FieldTypeArrayConvertor extends ArrayConverter
    {

      public FieldTypeArrayConvertor()
      {
        super(FIELD_TYPE[].class, new FieldTypeConvertor());
      }
    }

    ConvertUtils.register(new FieldTypeConvertor(), FIELD_TYPE.class);
    ConvertUtils.register(new FieldTypeConvertor(), FIELD_TYPE.class);
    ConvertUtils.register(new FieldTypeArrayConvertor(), FIELD_TYPE[].class);

    ConvertUtils.lookup(FIELD_TYPE.class).getClass();
  }

}