Hive UDFs — User-Defined Functions for Custom Logic in SQL

Three families, three plan shapes

The split between UDF, UDAF and UDTF is not a naming convention. Each family lands in a different place in the compiled plan, and that placement is what determines where you may use it and what the optimizer is permitted to do around it.

A scalar UDF is an expression. It becomes a node in an expression tree hanging off a Select, Filter or Join operator, and it is evaluated once per row per occurrence. Cardinality is unchanged, so the planner treats it as a pure value: it can fold the call away when the arguments are literals, evaluate a repeated subexpression once, or move a filter across it.

A UDAF is not an expression at all - it is a plan shape. The compiler turns SELECT k, myagg(v) FROM t GROUP BY k into two aggregation stages with a shuffle between them: a partial aggregation that runs where the data is, and a final aggregation that runs after rows have been redistributed by k. The intermediate result travels over the network. Everything that is hard about writing a UDAF follows from that one fact.

A UDTF is its own operator. It consumes one input row and pushes zero or more rows downstream through a forward() callback, so it changes cardinality mid-plan. Hive charges for that with SQL restrictions that regularly surprise people: a table function cannot sit beside other expressions in the same select list, cannot be nested inside another table function, and cannot share a query block with GROUP BY, CLUSTER BY, DISTRIBUTE BY or SORT BY. The escape hatch is LATERAL VIEW, which gives the table function its own operator and correlates its output back to the originating row. LATERAL VIEW OUTER keeps that row when the function emits nothing, which is the difference between inner and outer semantics and the reason a naive explode() quietly deletes rows with empty arrays.

The legacy UDF class and why GenericUDF replaced it

The original API is org.apache.hadoop.hive.ql.exec.UDF. You extend it, write one or more methods literally named evaluate, and Hive selects the overload reflectively from the argument types at compile time. It is still supported, still what a fifteen-year-old internal JAR is built against, and a dead end for new work, for four structural reasons.

Reflective invocation. The resolved method is called through the reflection API on every row, which means arguments are marshalled into an Object[] and primitives are boxed, per row, forever. Fixed arity and fixed types. The set of overloads you compiled is the set of signatures you support; you cannot write a function whose return type is a function of its argument types, nor one that accepts a variable number of arguments. No usable complex types. The signature speaks in Java and Writable primitives, so structs, arrays and maps have nowhere to go. Eager arguments. Every argument is evaluated before your method is entered, so a conditional function cannot skip the branch it does not need.

GenericUDF exists to fix all four, and its three methods map onto the three moments in a query's life:

ObjectInspector initialize(ObjectInspector[] arguments) runs once, when the operator is set up rather than when rows arrive. You check the argument count and categories here and throw UDFArgumentLengthException or UDFArgumentTypeException so the query fails at compile time instead of on row one of a four-hour job. You then return the ObjectInspector that describes your result - which is how a function computes its own return type, deciding that a decimal input of a given precision produces a decimal output of a wider one.

Object evaluate(DeferredObject[] arguments) runs per row, and the DeferredObject wrapper is the lazy evaluation the legacy API lacked. Each element holds an unevaluated child expression; calling get() is what forces it. A conditional function therefore forces only the branch it takes, so an expensive argument on the untaken side costs nothing. getDisplayString() supplies the text Hive prints for your call in EXPLAIN output.

public class MaskEmail extends GenericUDF {

  private StringObjectInspector in;
  private final Text out = new Text();      // reused: never allocate per row

  @Override
  public ObjectInspector initialize(ObjectInspector[] args)
      throws UDFArgumentException {
    if (args.length != 1) {
      throw new UDFArgumentLengthException("mask_email expects 1 argument");
    }
    if (!(args[0] instanceof StringObjectInspector)) {
      throw new UDFArgumentTypeException(0, "mask_email expects a string");
    }
    in = (StringObjectInspector) args[0];
    return PrimitiveObjectInspectorFactory.writableStringObjectInspector;
  }

  @Override
  public Object evaluate(DeferredObject[] args) throws HiveException {
    Object arg = args[0].get();             // forced here, not before
    if (arg == null) {
      return null;
    }
    String s = in.getPrimitiveJavaObject(arg);
    int at = s.indexOf('@');
    out.set(at < 2 ? "***" : s.charAt(0) + "***" + s.substring(at));
    return out;                             // matches the OI declared above
  }

  @Override
  public String getDisplayString(String[] children) {
    return "mask_email(" + children[0] + ")";
  }
}

ObjectInspectors - passing values without boxing them

An ObjectInspector is a separate object that knows how to read a value. The value itself crosses the API boundary as a bare Object, and the inspector is handed to you once, during initialize(). That pairing is the whole trick: type dispatch happens a single time per column per operator rather than once per row, and the value is free to stay in whatever representation the reader produced.

This matters because Hive's row formats are deliberately lazy. A row from a text table is a byte buffer with field offsets, not a set of Java objects; a row out of ORC arrives as Writables backed by reusable buffers. A lazy string inspector can hand you the field without materialising anything. Only when you call getPrimitiveJavaObject() do you pay for a real java.lang.String; getPrimitiveWritableObject() keeps you on the Writable, which is what you want when you are only going to copy bytes onward. Inspectors are grouped into categories - primitive, list, map, struct, union - and the structured ones expose accessors that walk into a value by field or index without converting the parts you never touch.

Three rules keep this from turning into runtime casts. First, whatever OI you returned from initialize() is a promise, and the object you return from evaluate() must honour it; advertising the writable string inspector and then returning a java.lang.String is the single most common source of a ClassCastException raised somewhere in a serializer with no reference to your class in the message. Second, use the converter utilities rather than hand-writing a type matrix when a function should accept anything numeric. Third - and this one bites hardest inside aggregate functions - if you intend to keep a value past the current row, copy it into a standard object first. The instance you were handed is very often reused for the next row, and stashing the reference gives you a collection containing the last row repeated N times.

The UDAF lifecycle and the merge contract

A user-defined aggregate is two classes. A resolver, typically extending AbstractGenericUDAFResolver, is asked at compile time to supply an evaluator for a given set of argument types. The evaluator, a GenericUDAFEvaluator, does the work, and it has more entry points than people expect: init(), getNewAggregationBuffer(), reset(), iterate(), terminatePartial(), merge() and terminate(). The buffer is your per-group state; reset() exists so the engine can recycle one buffer across many groups instead of allocating per group.

The reason init() receives a Mode is that one evaluator class runs in up to four distinct configurations, and which of the seven methods are actually called depends entirely on which one it was given:

COMPLETE is the mode that hides the bug. On a small input, or when the grouping column already matches how the data is laid out, or in a single-stage plan, Hive can run the aggregate end to end in one place: iterate then terminate, never calling merge or terminatePartial at all. Your unit test passes. The query you developed against last month's partition passes. Then the input grows, the compiler picks PARTIAL1 followed by FINAL, and the code path nobody exercised runs in production. It does not throw. It returns a number that looks like an answer.

Two invariants keep you out of that. The state emitted by terminatePartial must be sufficient to reconstruct the aggregate from arbitrary subsets - an average ships a sum and a count, never a running mean; a distinct count ships a sketch, never the count so far. And the state must be bounded. An evaluator whose buffer is a list of every value it has observed is not an aggregate, it is a collector: it works until the first large group, then it takes the task's heap with it. Where you genuinely need all the values - a median, a percentile - the answer is a bounded summary that merges associatively, not a bigger heap.

Determinism, statefulness, and what the optimizer may do

The @UDFType annotation on the class is a contract you are making with the compiler. Its two load-bearing attributes are deterministic, which defaults to true, and stateful, which defaults to false. Nothing verifies either one; they are assertions, and the optimizer acts on them.

Declared deterministic, a function may be constant folded: if constant propagation reduces every argument to a literal, the compiler is free to invoke the function once while planning and substitute the result into the plan. It may be evaluated once for repeated subexpressions within a row, which is what hive.cache.expr.evaluation governs. And it may be moved relative to filters and joins - a deterministic predicate can be pushed toward the scan (see predicate pushdown), where it will be applied to a different set of rows than the SQL text implies.

That last clause is why a mislabelled function produces non-reproducible results rather than an error. A function that reads the clock, consults a cache, or picks up a row-dependent seed, but is left with the default annotation, can be folded to a single compile-time value that every output row then carries; or hoisted across a join so it runs on a different row multiset; or skipped for a duplicate occurrence whose value should have differed. Nothing fails. The query simply returns a different answer on Tuesday than it did on Monday, and the discrepancy surfaces weeks later in a reconciliation.

stateful = true is a stronger claim than merely non-deterministic: it says invocation N depends on the invocations before it, so the engine must not reorder calls, elide them, or change how many happen. A sequence generator is the canonical case. Annotate honestly and the optimizer works around you; annotate optimistically and it optimises through you.

The vectorization tax

A GenericUDF has no vector implementation unless somebody wrote one, so Hive falls back to VectorUDFAdaptor: it unpacks the column batch into rows, calls your function once per row, and repacks the results into an output column vector. That is better than dropping the operator to row mode outright, but it reintroduces precisely the per-row object churn that vectorized execution exists to remove, and when the expression or type combination cannot be adapted the operator falls back completely. hive.vectorized.adaptor.usage.mode is the policy knob - none, chosen (the default, restricting the adaptor to functions Hive has vetted) and all - and EXPLAIN VECTORIZATION DETAIL names the operator and the reason it stayed behind.

Batch layout, column vectors and how to write a vectorized expression belong to Hive vectorization. The equivalent problem on the other engine - why a Java function cannot be inlined into generated machine code, and why Impala accepts only legacy-style Java UDFs and no Java aggregates at all, with native functions as the fast path - belongs to Impala code generation. The design rule that lives here is narrower: a scalar function on a hot path either gets a vector implementation or gets benchmarked with the adaptor overhead counted in.

Temporary, permanent, and the JAR distribution problem

Registration comes in two scopes. A temporary function is created against the current session after an ADD JAR, lives in that session's classloader, and vanishes when the session ends - convenient for development, invisible to anyone else. A permanent function is a row in the metastore, namespaced to a database and carrying the URI of its JAR, so every session against that metastore can see it.

-- session scope: the JAR must be reachable from the HiveServer2 process
ADD JAR /opt/udf/text-tools-1.4.0.jar;
CREATE TEMPORARY FUNCTION mask_email AS 'com.example.hive.MaskEmail';

-- cluster scope: the JAR must be reachable from every node that runs a task
CREATE FUNCTION analytics.mask_email
    AS 'com.example.hive.MaskEmail'
    USING JAR 'hdfs:///apps/hive/udf/text-tools-1.4.0.jar';

RELOAD FUNCTIONS;                       -- other HiveServer2 instances refresh
DESCRIBE FUNCTION EXTENDED analytics.mask_email;

-- redeploy by version, never by overwriting the file in place
DROP FUNCTION analytics.mask_email;
CREATE FUNCTION analytics.mask_email
    AS 'com.example.hive.MaskEmail'
    USING JAR 'hdfs:///apps/hive/udf/text-tools-1.5.0.jar';

The class has to be loadable in two very different places: the HiveServer2 JVM that compiles the query, and every container or daemon that executes a fragment of it. For a permanent function the registered URI is what makes the second half work - the JAR is localised from that location onto the executing node. Which is why a file:/// path that happens to exist on the HiveServer2 host is the classic "works in Beeline, ClassNotFoundException in the task" report: the compile-time lookup succeeded and the runtime one had nothing to fetch.

Two further sharp edges. Function registries are cached per HiveServer2 instance, so a function created through one endpoint is not necessarily visible through another until it refreshes; RELOAD FUNCTIONS is the manual push. And your JAR is loaded into a JVM that already has Hive's own dependency set on its classpath, without an isolating classloader. If you need a different version of a common library than the cluster ships, shade and relocate it at build time. The alternative is discovering at runtime which copy won, usually through a NoSuchMethodError on a method that plainly exists in the version you compiled against.

Where custom functions take the heap with them

Custom code is a leading cause of Hive task failures, and the reasons repeat.

Unbounded aggregation state. Map-side partial aggregation, enabled by hive.map.aggr, keeps a hash map from group key to aggregation buffer inside the task heap, and Hive polices its size against a configured fraction of that heap, flushing when the estimate crosses the line. The estimate is only as good as the size it can compute for your buffer. A buffer holding a growing collection is opaque to that accounting, so the task sails past the threshold and dies with a heap error whose message names a Java collection class rather than your aggregate.

Per-row allocation. A function invoked two billion times that allocates a date formatter, a compiled pattern, a result object and two substrings on each call generates ten billion short-lived objects. This rarely presents as an OOM; it presents as a task burning a third of its CPU in young-generation collection with no obvious culprit. Compile patterns and formatters once in initialize() and hold them as instance fields, reuse a single Writable for the return value, and keep string concatenation out of the inner loop.

External calls. A function that contacts a service per row multiplies that service's latency by your row count, pins a container for the duration, and turns an ordinary speculative or retried task into duplicate traffic on the far side. Enrichment almost always wants to be a join - broadcast the lookup table into a map-side join - not a per-row request.

Undecided error handling. A function that throws on one malformed value fails its task, gets retried, fails again, and kills a query that was 90% done because of a single bad row in a billion. Choose deliberately: either return null for unparseable input and surface the count as a metric, or fail loudly and immediately. The failure mode you did not choose is the one you will get.

Testing and benchmarking one

A GenericUDF is an ordinary Java object, so it needs no cluster to test. Construct it, call initialize() with inspectors from the primitive factory, and drive evaluate() with deferred wrappers around plain Java values.

@Test
public void masksTheLocalPart() throws Exception {
  MaskEmail udf = new MaskEmail();
  udf.initialize(new ObjectInspector[] {
      PrimitiveObjectInspectorFactory.javaStringObjectInspector });

  Object r = udf.evaluate(new DeferredObject[] {
      new GenericUDF.DeferredJavaObject("sandeep@example.com") });
  assertEquals("s***@example.com", r.toString());

  assertNull(udf.evaluate(new DeferredObject[] {
      new GenericUDF.DeferredJavaObject(null) }));
}

Three cases catch most of what goes wrong. A null in every argument position, because null handling is the bug that reaches production most often. A wrongly typed argument, asserting that initialize() is what rejects it - if the failure only appears from evaluate(), your type checking is running per row and your users get a runtime error instead of a compile error. And, for an aggregate, an explicit two-phase test: drive one evaluator instance in PARTIAL1 over one slice of input, a second over another slice, collect both partial results, then feed them to a third instance initialised in FINAL and assert the answer matches a single-pass computation over the union. That test is the one nobody writes, and it is the only one that catches a broken merge before the data volume does.

Benchmarking starts with per-call cost in isolation - a microbenchmark harness gives you nanoseconds per invocation, and the arithmetic from there is unforgiving: 200ns across five billion rows is roughly 17 minutes of pure CPU, before the engine does anything else. Confirm it in the engine by timing the same query with and without the function forced into evaluation, and read EXPLAIN VECTORIZATION on both plans. If the version with your function has dropped off the vectorized path, the microbenchmark understated the cost, because you are now also paying for every other expression in that operator to run a row at a time.