Pig UDFs and scripting

Apache Pig Tutorial 1edureka! · Pig overviewUser Defined Functions in PIG Using JAVAJank Man · Java UDFs in PigPig UDFsAcademy of Fine Learning · Pig UDFsDemo on Pig UDFedureka! · Pig UDF demo

Why legacy Java belongs in PigSearch on YouTubeBookmarkSaved

Real teams already own tested Java logic: tax-ID checks, credit scores, locale-specific parsers. Rewriting that logic in Pig Latin risks new bugs. It also duplicates maintenance work.

A UDF runs that Java class unchanged inside a Pig Latin pipeline. Pig gives the dataflow. Java gives the exact logic. Typical UDF candidates from the slides:

UDF kindsSearch on YouTubeBookmarkSaved

A UDF extends Pig Latin with logic that built-in operators lack. You write it once. Then you call it like any built-in function.

A filter function is a special Eval function. It returns true or false. An aggregate function can use the algebraic interface. That interface lets Pig use a combiner for speed.

The integration pathSearch on YouTubeBookmarkSaved

Legacy code reaches Pig Latin in five steps. The slides show this path with a tax-ID validator.

The slide example registers a legacy tax-ID checker:

REGISTER /usr/local/pig/legacy_udf.jar;
DEFINE validate com.company.legacy.TaxIdValidator();
clean = FILTER records BY validate(tax_id) == true;

Writing a Java UDFSearch on YouTubeBookmarkSaved

Every Java Eval UDF extends org.apache.pig.EvalFunc<T>. It overrides one method: exec(Tuple input). The input tuple holds the call arguments. The return value of exec() becomes the new field value.

package myudfs;
import java.io.IOException;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;

public class UPPER extends EvalFunc<String> {
  public String exec(Tuple input) throws IOException {
    if (input == null || input.size() == 0) return null;
    try {
      String str = (String) input.get(0);
      return str.toUpperCase();
    } catch (Exception e) {
      throw new IOException("Caught exception processing input row", e);
    }
  }
}

Three rules from the slides. Return null on missing input. Catch bad rows and rethrow as IOException. Never let a raw exception escape.

REGISTER myudfs.jar;
DEFINE UPPER myudfs.UPPER();
upper_names = FOREACH students GENERATE UPPER(name);

REGISTER and DEFINESearch on YouTubeBookmarkSaved

REGISTER makes a jar visible. Without an alias, you call the full class name:

REGISTER myudfs.jar;
upper_names = FOREACH students GENERATE myudfs.UPPER(name);

DEFINE gives a short name. Use it when a script calls the UDF often:

REGISTER myudfs.jar;
DEFINE UPPER myudfs.UPPER();
upper_names = FOREACH students GENERATE UPPER(name);

DEFINE can also pass constructor arguments. This configures the UDF at load time. The slides name this use but give no fixed syntax. The form is DEFINE alias package.Class(arg1, arg2);. DEFINE can also give a STREAM command a short alias.

Piggybank and UDF librariesSearch on YouTubeBookmarkSaved

Piggybank is a community library of ready UDFs. It ships with Pig. It covers strings, math and dates. Use it instead of writing common logic again.

REGISTER '/path/to/piggybank.jar';
DEFINE ISOToUnix org.apache.pig.piggybank.evaluation.datetime.convert.ISOToUnix();

Other libraries from the slides:

Scripting UDFsSearch on YouTubeBookmarkSaved

A UDF can use a scripting language. Pig runs it through an embedded interpreter. This suits quick one-off logic. No compile step. No jar to manage.

# myfuncs.py
@outputSchema("upper:chararray")
def upper(s):
    return s.upper()

REGISTER 'myfuncs.py' USING jython AS myfuncs;
upper_names = FOREACH students GENERATE myfuncs.upper(name);

The @outputSchema line is optional. Without it Pig types the result as bytearray. Add it here to get chararray. Supported languages from the slides:

The MAPREDUCE operatorSearch on YouTubeBookmarkSaved

Many teams own compiled MapReduce jobs. Rewriting them in Pig Latin is not realistic. The MAPREDUCE operator runs one such jar as a single step inside a Pig pipeline. Pig Latin runs before and after it.

Pig moves the data both ways. It STOREs a relation to the path the jar expects. It LOADs the jar output back with a schema you declare. The jar itself has no schema.

result = MAPREDUCE 'legacy-job.jar'
  STORE cleaned_orders INTO 'mr-input'
  LOAD 'mr-output' AS (customer_id:int, score:double)
  `com.example.LegacyScoringJob mr-input mr-output`;

Developing Pig Latin scriptsSearch on YouTubeBookmarkSaved

The slides give a four-step workflow. Follow it in order.

1. Prototype in local mode with pig -x local. Use a small sample file. Local runs skip cluster queues.
2. Build incrementally. Add one relation at a time. DUMP or DESCRIBE it before you write the next line.
3. Name aliases well (valid_orders, not a). Add -- comments. A teammate must read the script without you.
4. Scale up only when the logic is right. Switch to pig -x mapreduce (or Tez/Spark). Point LOAD at the full employee.csv file.

pig -x local filter_high_earners.pig
pig -x mapreduce filter_high_earners.pig

Three Python patternsSearch on YouTubeBookmarkSaved

Plain Pig Latin has no loops, no conditionals and no real variables. It has only $-parameter substitution. Python adds all three. The slides give three patterns.

Only the third pattern puts Python logic around each run. The first two treat Pig as a black box. Rule of thumb from the slides: embed only when you need a loop or a condition. The first two patterns suit all else.

Generating Pig Latin from PythonSearch on YouTubeBookmarkSaved

Python builds the .pig file text from a template. An f-string fills in the threshold and paths. The output is plain Pig Latin.

# generate_pig.py
SALARY_THRESHOLD = 50000
INPUT_FILE = 'employee.csv'
OUTPUT_DIR = 'high_earners_output'

script = f"""
emp = LOAD '{INPUT_FILE}' USING PigStorage(',')
  AS (id:int, name:chararray, dept:chararray,
      salary:int, age:int, city:chararray);
high_earners = FILTER emp BY salary > {SALARY_THRESHOLD};
STORE high_earners INTO '{OUTPUT_DIR}';
"""

with open('filter_high_earners.pig', 'w') as f:
    f.write(script)
print('Generated filter_high_earners.pig')

Run it with python generate_pig.py. Then open filter_high_earners.pig and check the text. The generated script is ordinary Pig Latin. Something else must launch it.

Running Pig as a subprocessSearch on YouTubeBookmarkSaved

subprocess.run() launches the pig command like a terminal would. Capture the output. Check the return code. React to failure in code.

# run_pig.py
import subprocess

result = subprocess.run(
    ['pig', '-x', 'local', 'filter_high_earners.pig'],
    capture_output=True, text=True
)

if result.returncode == 0:
    print('Pig job succeeded')
    print(result.stdout)
else:
    print('Pig job FAILED')
    print(result.stderr)
    raise SystemExit(1)

Try it now: save the file, run python run_pig.py, then break the .pig file on purpose and re-run. The second run must print Pig job FAILED with a nonzero exit code. You get pass/fail plus logs. You get no mid-run control.

Embedding Pig Latin in PythonSearch on YouTubeBookmarkSaved

Embedding differs from a scripting UDF. A scripting UDF writes one function. Embedding drives the whole script from Python. Python runs under Jython inside Pig's own JVM. Run the file with pig, not python.

from org.apache.pig.scripting import Pig

P = Pig.compile("emp = LOAD '$input' USING PigStorage(',') AS (...); ...")
bound = P.bind({'input': 'students.txt', 'city': 'Delhi', 'output': 'out'})
stats = bound.runSingle()

The slide example loops over departments. compile() runs once. bind() runs once per department with fresh $DEPT and $OUT. Each run writes one output folder.

from org.apache.pig.scripting import Pig

pig_latin = """
emp = LOAD 'employee.csv' USING PigStorage(',')
  AS (id:int, name:chararray, dept:chararray,
      salary:int, age:int, city:chararray);
mine = FILTER emp BY dept == '$DEPT';
summary = FOREACH (GROUP mine ALL) GENERATE COUNT(mine), AVG(mine.salary);
STORE summary INTO '$OUT';
"""

P = Pig.compile(pig_latin)
for dept in ['Engineering', 'Sales', 'Marketing']:
    out_dir = dept.lower() + '_avg_output'
    bound = P.bind({'DEPT': dept, 'OUT': out_dir})
    stats = bound.runSingle()
    status = 'OK' if stats.isSuccessful() else 'FAILED'
    print(dept, '->', status)

Run it with pig embed_dept_avg.py. Confirm three output folders appear.

Testing and debugging scriptsSearch on YouTubeBookmarkSaved

DESCRIBE, EXPLAIN and ILLUSTRATE check a script while you build it. PigUnit tests it repeatably without a cluster.

Manual checks suit interactive work in Grunt. Nobody re-runs them before each deploy. PigUnit runs in a normal build and catches a broken FILTER or JOIN key early.

DUMP, DESCRIBE and ILLUSTRATESearch on YouTubeBookmarkSaved

DUMP prints a relation to the console. Every DUMP starts a real mini MapReduce job, even in local mode. Use it on purpose, not after each keystroke. Never DUMP a huge relation. Limit it first.

-- cheap: schema only, no job runs
DESCRIBE emp;
-- safe peek at big data: only 5 rows
DUMP (LIMIT emp 5);
-- done: write the full result
STORE dept_summary INTO 'dept_summary_output' USING PigStorage(',');

Schema and field errorsSearch on YouTubeBookmarkSaved

When Pig complains, run DESCRIBE on every relation above the failing line first. Fix the schema before you touch logic.

STORE and existing directoriesSearch on YouTubeBookmarkSaved

STORE writes final output to the file system. It does not print to the console. Pig refuses to overwrite an existing output directory. This guards results against silent loss. On re-run you see:

Output Location Validation Failed: 'dept_summary_output' already exists

Three fixes from the slides:

hadoop fs -rm -r dept_summary_output

Or use a fresh path each run (dept_summary_output_v2). Or wrap both steps in a shell script: delete with hadoop fs -rm -r -f out, then run pig myscript.pig.

PigUnit testsSearch on YouTubeBookmarkSaved

PigUnit is JUnit-based. It swaps real LOAD/STORE for in-memory rows. It runs the script and asserts on the named output relation.

PigTest test = new PigTest("script.pig");

String[] input = {
   "1\tRajiv\tHyderabad",
   "2\tOmer\tKolkata"
};

String[] output = {
   "(1,RAJIV,Hyderabad)",
   "(2,OMER,Kolkata)"
};

test.assertOutput("students", input, "upper_names", output);

assertOutput() feeds input into the students alias. It runs the script through upper_names. It fails the test on any mismatch. No file system or cluster takes part.

Employee analytics labSearch on YouTubeBookmarkSaved

This lab joins all ideas of the lecture: load, filter, project, group, aggregate, store, then automate with Python.

The employee dataSearch on YouTubeBookmarkSaved

employee.csv has 10 rows with schema (id:int, name:chararray, dept:chararray, salary:int, age:int, city:chararray). Push it to HDFS first on a real cluster:

hadoop fs -mkdir -p /pig_lab
hadoop fs -put employee.csv /pig_lab/employee.csv

Task listSearch on YouTubeBookmarkSaved

1. LOAD the file with an explicit schema into emp.
2. DUMP and DESCRIBE emp. Confirm 10 rows and correct types.
3. FILTER salaries above 50000 into high_earners. Confirm 5 rows.
4. FOREACH high_earners to keep name, dept, salary.
5. GROUP emp by dept. Confirm 3 groups.
6. One FOREACH with COUNT, AVG and MAX per group.
7. STORE the result into dept_summary_output. List it with hadoop fs -ls dept_summary_output (in local mode, use ls dept_summary_output). The output directory holds part-files inside.
8. Re-run the STORE on purpose. Reproduce the exists error. Fix it.

Reference solutionSearch on YouTubeBookmarkSaved

emp = LOAD 'employee.csv' USING PigStorage(',')
  AS (id:int, name:chararray, dept:chararray,
      salary:int, age:int, city:chararray);

high_earners = FILTER emp BY salary > 50000;

emp_slim = FOREACH high_earners GENERATE name, dept, salary;

grouped = GROUP emp BY dept;

dept_summary = FOREACH grouped GENERATE
  group AS department,
  COUNT(emp) AS emp_count,
  AVG(emp.salary) AS avg_salary,
  MAX(emp.salary) AS max_salary;

STORE dept_summary INTO 'dept_summary_output' USING PigStorage(',');

Bonus: configurable generator (extra)Search on YouTubeBookmarkSaved

Hard-code nothing. Take the department and threshold from the command line. Write one .pig file per department so reruns do not clash.

import argparse

parser = argparse.ArgumentParser()
parser.add_argument('--dept', required=True)
parser.add_argument('--min-salary', type=int, default=50000)
args = parser.parse_args()
out_file = f'{args.dept.lower()}_filtered.pig'

script = f"""
emp = LOAD 'employee.csv' USING PigStorage(',')
  AS (id:int, name:chararray, dept:chararray,
      salary:int, age:int, city:chararray);
result = FILTER emp BY dept == '{args.dept}' AND salary > {args.min_salary};
DUMP result;
"""

with open(out_file, 'w') as f: f.write(script)

Run python generate_pig.py --dept Engineering --min-salary 60000. The file engineering_filtered.pig keeps Engineering rows above 60000: Priya (62000) and Divya (71000). Rajiv (55000) and Rakesh (48000) drop out.

Quick self-testSearch on YouTubeBookmarkSaved