MyException - 我的异常网
当前位置:我的异常网» 开源软件 » sqoop扩张

sqoop扩张

www.MyException.Cn  网友分享于:2013-02-21  浏览:123次
sqoop扩展
sqoop是一款数据互导工具,利用它可以在关系数据库与hbase/hive/hdfs进行数据互导,过程中会生成一个mapreduce任务,也就说是sqoop是基于hadoop的.淘宝也有自己的数据互导工具,叫datax,它跟sqoop实现原理不同,没有基于hadoop.
 
sqoop目前还没有很好的支持不同的导数据业务,比如数据库字段名称不能与java关键字同名、row-key不支持整型导入等,所以还需要对sqoop进行扩展,使其能很好地支持我们的业务。今天实现了row-key支持整形导入
 
继承PutTransformer类
public class ToIntPutTransformer extends PutTransformer {

  public static final Log LOG = LogFactory.getLog(
      ToIntPutTransformer.class.getName());

  // A mapping from field name -> bytes for that field name.
  // Used to cache serialization work done for fields names.
  private Map<String, byte[]> serializedFieldNames;

  public ToIntPutTransformer() {
    serializedFieldNames = new TreeMap<String, byte[]>();
  }

  /**
   * Return the serialized bytes for a field name, using
   * the cache if it's already in there.
   */
  private byte [] getFieldNameBytes(String fieldName) {
    byte [] cachedName = serializedFieldNames.get(fieldName);
    if (null != cachedName) {
      // Cache hit. We're done.
      return cachedName;
    }

    // Do the serialization and memoize the result.
    byte [] nameBytes = Bytes.toBytes(fieldName);
    serializedFieldNames.put(fieldName, nameBytes);
    return nameBytes;
  }

  @Override
  /** {@inheritDoc} */
  public List<Put> getPutCommand(Map<String, Object> fields)
      throws IOException {

    String rowKeyCol = getRowKeyColumn();
    String colFamily = getColumnFamily();
    byte [] colFamilyBytes = Bytes.toBytes(colFamily);

    Object rowKey = fields.get(rowKeyCol);
    if (null == rowKey) {
      // If the row-key column is null, we don't insert this row.
      LOG.warn("Could not insert row with null value for row-key column: "
          + rowKeyCol);
      return null;
    }

    //rowKey转换为整形,指定的rowKey对应的数据库字段要是整型,否则报错
    Put put = new Put(Bytes.toBytes(Integer.parseInt(rowKey.toString())));

    for (Map.Entry<String, Object> fieldEntry : fields.entrySet()) {
      String colName = fieldEntry.getKey();
      if (!colName.equals(rowKeyCol)) {
        // This is a regular field, not the row key.
        // Add it if it's not null.
        Object val = fieldEntry.getValue();
        if (null != val) {
          put.add(colFamilyBytes, getFieldNameBytes(colName),
              Bytes.toBytes(val.toString()));
        }
      }
    }

    return Collections.singletonList(put);
  }
 编译打包,放到lib目录下.
配置conf/sqoop-site.xml文件,增加
<property>
    <name>sqoop.hbase.insert.put.transformer.class</name>
    <value>org.apache.sqoop.hbase.ToIntPutTransformer</value>
</property>
 完成.

文章评论

软件开发程序错误异常ExceptionCopyright © 2009-2015 MyException 版权所有