自定义UDF函数,从hive保存到mongodb

(可以通过idea工具调试UDF函数,第二步中会提供参考)

一、自定义UDF函数:

1、首先是pom.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.cn.hive.mongo</groupId>
    <artifactId>hive-mongodb</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>hive-mongodb</name>
    <description>load data hive to mongodb</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.7</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>0.14.0</version>
            <type>jar</type>
            <scope>provided</scope>
            <exclusions>
                <exclusion>
                    <artifactId>calcite-core</artifactId>
                    <groupId>org.apache.calcite</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>calcite-avatica</artifactId>
                    <groupId>org.apache.calcite</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>mongo-java-driver</artifactId>
            <version>3.4.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-core</artifactId>
            <version>0.20.205.0</version>
            <type>jar</type>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.mongodb</groupId>
            <artifactId>bson</artifactId>
            <version>3.2.2</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <configuration>
                    <archive>
                        <manifest>
                            <addClasspath>true</addClasspath>
                            <mainClass>com.cn.udf.MongoStorageHandler</mainClass>
                        </manifest>
                    </archive>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

2、关于mongo的Helper类

package com.cn.udf;

import com.mongodb.*;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateOptions;
import org.bson.Document;


/**
 * Created by Administrator on 2017/1/13 11:30.
 * Copyright: Copyright (c) 2016 
 * Description:
 */
public class MongodbHelper {

    private static MongoClient mongoClient = null;
    private static MongoDatabase database = null;

    public MongodbHelper() {

    }

    public MongoDatabase init(String dbUri) {
        String dbName = dbUri.substring(dbUri.lastIndexOf("/") + 1, dbUri.length());
        if (database == null) {
            MongoClientURI uri = new MongoClientURI(dbUri,
                    MongoClientOptions.builder().cursorFinalizerEnabled(false));
            mongoClient = new MongoClient(uri);
            database = mongoClient.getDatabase(dbName);
        }
        return database;
    }


    public MongoCollection<Document> getCollection(String dbCollection) {
        MongoCollection<Document> collection;
        collection = database.getCollection(dbCollection);
        return collection;
    }

    public void close() {
        mongoClient.close();
    }

    public void updateOneDocument(MongoCollection<Document> collection, String[] arguments) {
        FindIterable<Document> documents = collection.find(Filters.eq("Id", arguments[0]));
        Document document = documents.first();
        if (null == document) {
            Document docs = new Document().append("Id", arguments[0]).append("count", new Integer(arguments[1]));
            collection.insertOne(docs);
        } else {
            Long last_time_hive = new Long(arguments[2]);
            Long create_mongo = new Long(String.valueOf(document.get("startStamp")));
            String phone = document.get("phone") == null ? "" : String.valueOf(document.get("phone"));
            if (last_time_hive <= create_mongo) {
                collection.updateMany(Filters.and(Filters.eq("Id", arguments[0])),
                        new Document("$inc", new Document("count", new Integer(arguments[1]) + 1))
                                .append("$set", new Document("phone", ("".equals(phone)) ? arguments[3] : phone)),
                        new UpdateOptions().upsert(true));
            } 
        }
    }

}

3、自定义函数MongoStorageHandler入口类

package com.cn.udf;

import com.mongodb.client.MongoCollection;
import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.UDFType;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde.Constants;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.IntWritable;

/**
 * Created by Administrator on 2017/1/12 18:56.
 * Copyright: Copyright (c) 2016 
 * Description: load hive data to mongodb
 */
@Description(name = "hive-to-mongodb",
        value = "_FUNC_(url, collection, [arguments]) - sends data to a jdbc driver\n"
                + "-- url : mongodb://host:port/dbName or mongodb://username/password@host:port/dbName\n"
                + "-- collection : database collection\n"
                + "-- [arguments] order : customer_id, sign_times, sign_last_time, phone, sex, name\n"
                + "-- return 0 for success, -1 for failed, 2 for connection problem",
        extended = "argument 0 is the database url string(mongodb://host:port/dbName or mongodb://username/password@host:port/dbName)\n"
                + "argument 1 is the database collection\n"
                + "argument (2~n) The remaining arguments must be primitive and are "
                + "passed to the PreparedStatement object\n"
                + "if return 2 stand for you have a problem with the Driver loading or connection\n")
@UDFType(deterministic = false)

public class MongoStorageHandler extends GenericUDF {

    private static final Logger log = LoggerFactory.getLogger(MongoStorageHandler.class);

    private transient ObjectInspector[] argumentOI;
    private final IntWritable result = new IntWritable(0);

    private MongodbHelper dbHelper = null;
    private MongoCollection<Document> connection = null;

    /**
     * Created by : Administrator . Created time: 2017/1/12 19:26.
     * Description: this should be connection;
     * Param Description: url,username,password,query,column1[,column2...]*
     */
    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
        argumentOI = arguments;
        for (int i = 0; i < 4; i++) {
            if (arguments[i].getCategory() == ObjectInspector.Category.PRIMITIVE) {
                PrimitiveObjectInspector poi = ((PrimitiveObjectInspector) arguments[i]);

                if (!(poi.getPrimitiveCategory() == PrimitiveObjectInspector.PrimitiveCategory.STRING)) {
                    throw new UDFArgumentTypeException(i,
                            "The argument of function should be \""
                                    + Constants.STRING_TYPE_NAME + "\", but \""
                                    + arguments[i].getTypeName()
                                    + "\" is found");
                }
            }
        }
        for (int i = 4; i < arguments.length; i++) {
            if (arguments[i].getCategory() != ObjectInspector.Category.PRIMITIVE) {
                throw new UDFArgumentTypeException(i,
                        "The argument of function should be primative"
                                + ", but \"" + arguments[i].getTypeName()
                                + "\" is found");
            }
        }

        return PrimitiveObjectInspectorFactory.writableIntObjectInspector;
    }

    /**
     * Created by : Administrator . Created time: 2017/1/12 19:28.
     * Description: 0 stand for success , -1 stand for failure
     */
    @Override
    public Object evaluate(DeferredObject[] arguments) throws HiveException {
        String url = arguments[0].get().toString();
        String dbCollection = arguments[1].get().toString();
        try {
            if (null == dbHelper){
                // 初始化链接
                dbHelper = new MongodbHelper();
            }
            dbHelper.init(url);
            connection = dbHelper.getCollection(dbCollection);
        } catch (Exception ex) {
            log.error("Driver loading or connection issue", ex);
            result.set(2);
        }

        if (connection != null) {
            try {
                // 逻辑部分
                String[] args = new String[10];
                for (int i = 2; i < arguments.length; ++i) {
                    args[i-2] = arguments[i].get() == null? "0" : arguments[i].get().toString();
                }
                dbHelper.updateOneDocument(connection, args);
            } catch (Exception e) {
                log.error("Underlying SQL exception", e);
                result.set(-1);
            }
        }
        return result;
    }

    @Override
    public String getDisplayString(String[] children) {
        StringBuilder sb = new StringBuilder();
        sb.append("dboutput(");
        if (children.length > 0) {
            sb.append(children[0]);
            for (int i = 1; i < children.length; i++) {
                sb.append(",");
                sb.append(children[i]);
            }
        }
        sb.append(")");
        return sb.toString();
    }

}

二、编写好UDF后,测试自定义函数:

准备工作:先将编写好的UDF函数jar包以及相关的依赖包上传服务器(windows下可选择WinSCP工具),比如:UDF函数上传到了/home/username/tmp目录下,依赖包在/home/username/tmp/lib下
1、在idea中配置Remote Configurations(如图一所示):

图 一 Remote配置Debug Configurations

2、启用hive的debug模式:
hive --debug

3、此时正在监听8000端口(Listening for transport dt_socket at address: 8000),在idea中启动配置好的Remote的Debug模式;
4、输入以下命令,自定义函数名(sign_func):

use testDb;
add jar /home/user/tmp/hive-mongodb-1.0-SNAPSHOT.jar;
create temporary function sign_func as 'com.cn.udf.MongoStorageHandler';
desc function sign_func;

5、调用自定义函数sign_func:

select sign_func('mongodb://your_host:your_port/your_dbName','your_collection',[column1,column2...]) from your_tableName ;
// 例如
select sign_func('mongodb://127.0.0.1:27017/test_db','test_coll',Id,col1,col2,col3) from tab_info limit 5;

OK,去mongodb查询一下数据是否存入mongo库。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,080评论 6 493
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,422评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,630评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,554评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,662评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,856评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,014评论 3 408
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,752评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,212评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,541评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,687评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,347评论 4 331
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,973评论 3 315
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,777评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,006评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,406评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,576评论 2 349

推荐阅读更多精彩内容

  • hive.ddl.output.format:hive的ddl语句的输出格式,默认是text,纯文本,还有json...
    博弈史密斯阅读 1,943评论 0 6
  • 你要写的查询有事无法轻松或是(无法完成)的查询任务,使用Hive内置函数来标识。 通过编写用户自定义函数(user...
    IT_小白阅读 726评论 0 1
  • hive简介 解释一:Hive是一个数据仓库基础工具在Hadoop中用来处理结构化数据。它架构在Hadoop之上,...
    卡卡xx阅读 6,304评论 0 4
  • Hive 内置函数 Hive自定义函数 UDF(User-Defined-Function) 一进一出 UDAF(...
    geekAppke阅读 3,572评论 0 0
  • 在hive中,有时候一些内置的函数,和普通的查询操作已经满足不了我们要查询的要求,这时候可以自己写一些自定义函数来...
    yannhuang阅读 2,443评论 0 0