使用 Clojure 编写 Hive UDF

Table of Contents

在 Hive 中可以用调用外部脚本的方式处理数据,用外部脚本的好处是不受语言限制,通常情况下我们用Python写一些外部脚本来处理一些数据,但是有些情况下写成 UDF(User-defined function)可以简化查询语句,也能更好地使用条件判断。

Hive 是用 Java 开发的,不过任何运行在 JVM 上的语言,只要支持和 Java 交互就可以写 UDF,选 Clojure 的原因很简单,因为它是 Lisp 方言,也能很好地和 Java 代码交互,并且开发效率很高。

Hive 的自定义函数分了三种:UDTF、UDAF 和 UDF,前两种我基本上用不上。

UDF 编写很简单,只要保证 Hive 能调用你暴露的 evaluate 函数(方法)即可,同时满足:

  1. 继承 org.apache.hadoop.hive.ql.exec.UDF;
  2. 调用 UDF 时,保证传入的参数和返回值为 org.apache.hadoop.io.Text 类型(参数和返回值也可以是 String 类型)

本文为 Hive 写一个 upper 函数,将字母转成大写(虽然很无聊,但因为代码量很少,容易让人看懂)

1 使用 Leiningen 创建项目

关于 Leiningen 是什么、怎么安装,这里不多说了。先创建个项目:

$ lein new upper

然后进入 upper 项目,修改 project.clj,如下:

(defproject upper "0.1.0-SNAPSHOT"
    :description "FIXME: write description"
    :url "http://example.com/FIXME"
    :license {:name "Eclipse Public License"
    :url "http://www.eclipse.org/legal/epl-v10.html"}
    :plugins [[cider/cider-nrepl "0.8.0-SNAPSHOT"]]
    :dependencies [[org.clojure/clojure "1.5.1"]
		   [hive/hive-exec "0.5.0"]
		   [org.apache.hadoop/hadoop-core "0.20.2-dev"]])

主要改动:

1、增加 Hadoop 和 Hive 的依赖

2、安装 cider/cider-nrepl 插件,便于和 Cider 交互开发

然后自动安装依赖:

$ lein deps

2 编写 UDF

编辑 src/upper/core.clj,详细请看代码注释:

(ns upper.core
  (:import [org.apache.hadoop.hive.ql.exec UDF])
  (:import [org.apache.hadoop.io Text])
  (:gen-class
   :name upper.core
   ;; 继承 org.apache.hadoop.hive.ql.exec.UDF,必须的
   :extends org.apache.hadoop.hive.ql.exec.UDF
   ;; 暴露 evaluate 方法给 Hive,同时声明参数类型和返回值类型
   ;; 列表里是参数类型,参数数目必须和列表数目对应
   ;; 参数和返回值都必须是 org.apache.hadoop.io.Text 类型
   :methods [[evaluate [org.apache.hadoop.io.Text] org.apache.hadoop.io.Text]]))

(defn upper [string]
  (.toUpperCase string))

;; #^Text 是元数据,声明函数返回的是 Text 类型
(defn #^Text -evaluate
  ;; 声明参数为 Text 类型
  [this #^Text string]
  ;; 先将参数值转成字符串类型处理后,再包装成 Text 类型
  (Text. (upper (.toString string))))

这里要注意一个细节,新建 Text 实例时不能给类型为 nil(对应 Java 中的 null)的参数,所以必须保证 upper 函数调用结果是字符串,否则会异常。

3 编译成 class 文件并打包

首先,需要修改 project.clj,增加 :aot 选项:

:aot [upper.core]

然后,将代码编译成 class。编译时一定要注意 JDK 版本,如果 Hive 用的 Java6,就必须用 Java6 编译:

$ lein compile Compiling upper.core

最后,如果编译顺利通过,再将它打包成 jar 文件:

$ lein uberjar Created /tmp/upper/target/upper-0.1.0-SNAPSHOT.jar Created /tmp/upper/target/upper-0.1.0-SNAPSHOT-standalone.jar

这时会在 target 目录下生成两个 .jar 文件,其中以“-standalone.jar”结尾的 jar 文比较大,应该有 17M 左右,因为它包含了所有的依赖包括 Clojure 自身),所以可以直接放 Hive 上使用。

4 在 Hive 中调用

为了方便测试,新建一个 hive.sql 文件,把 HiveQL 语句写进去:

-- 将生成的 jar 文件添加到 Hive 中,它会自动分发给其他节点
add jar target/upper-0.1.0-SNAPSHOT-standalone.jar;
-- 注册成 Hive 函数
create temporary function my_upper as 'upper.core';

select my_upper(en_name) from user_information;

然后执行:

$ hive -f hive.sql

如果顺利的话,Hive 将正常返回调用结果。

5 关于调试

开发 UDF 难免会遇到代码错误,可能是编译中出现错误,也可能是运行时出错。编译中出现错误一般就是语法一类的问题,很好解决;但如果是在 Hive 运行时出错,MapReduce 任务会被杀死,调试也会变得比较难,所以在这之前建议写好单元测试,保证各个部分的代码稳定后再放到 Hive 中。如果在 Hive 运行中死掉,可以到 JobTracker 的 Web 监控页面看调试日志中的 Java 异常信息。

6 关于性能优化

工作中我主要是写 UDF 处理海量日志,在写某个大量匹配功能的 UDF 时,一千万条日志花了 4 小时。但日志里要匹配的字段有大量是重复的,所以我用了 memoize 函数缓存结果,这时一千万条日志只跑了不到半个小时。如果你处理的数据中如果有比较多重复数据的话,建议使用 memoize。

另外,在大规模数据处理下,一些代码细节优化对速度提升很不明显,建议仔细揣摩代码,优先优化流程。像上面说的场景,我在优化了数据处理流程的情况下,最后只花几分钟就可以处理完一千万条日志了。