/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.pig.test.pigmix.datagen;
import java.io.*;
import java.lang.SecurityException;
import java.text.ParseException;
import java.util.*;
import sdsu.algorithms.data.Zipf;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.tools.cmdline.CmdLineParser;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;
/**
* A tool to generate data for performance testing.
*/
public class DataGenerator extends Configured implements Tool {
ColSpec[] colSpecs;
long seed = -1;
long numRows = -1;
int numMappers = -1;
String outputFile;
String inFile;
char separator = '\u0001' ;
Random rand;
private String[] mapkey = { "a", "b", "c", "d", "e", "f", "g", "h", "i",
"j", "k", "l", "m", "n", "o", "p", "q", "r", "s", "t", "u", "v", "w",
"x", "y", "z"};
public static void main(String[] args) throws Exception {
DataGenerator dg = new DataGenerator();
try {
ToolRunner.run(new Configuration(), dg, args);
}catch(Exception e) {
throw new IOException (e);
}
dg.go();
}
protected DataGenerator(long seed) {
System.out.println("Using seed " + seed);
rand = new Random(seed);
}
protected DataGenerator() {
}
protected DataGenerator(String[] args) {
}
public int run(String[] args) throws Exception {
CmdLineParser opts = new CmdLineParser(args);
opts.registerOpt('e', "seed", CmdLineParser.ValueExpected.REQUIRED);
opts.registerOpt('f', "file", CmdLineParser.ValueExpected.REQUIRED);
opts.registerOpt('r', "rows", CmdLineParser.ValueExpected.REQUIRED);
opts.registerOpt('s', "separator", CmdLineParser.ValueExpected.REQUIRED);
opts.registerOpt('i', "input", CmdLineParser.ValueExpected.REQUIRED);
opts.registerOpt('m', "mappers", CmdLineParser.ValueExpected.OPTIONAL);
char opt;
try {
while ((opt = opts.getNextOpt()) != CmdLineParser.EndOfOpts) {
switch (opt) {
case 'e':
seed = Long.valueOf(opts.getValStr());
break;
case 'f':
outputFile = opts.getValStr();
break;
case 'i':
inFile = opts.getValStr();
break;
case 'r':
numRows = Long.valueOf(opts.getValStr());
break;
case 's':
separator = opts.getValStr().charAt(0);
break;
case 'm':
numMappers = Integer.valueOf(opts.getValStr());
break;
default:
usage();
break;
}
}
} catch (ParseException pe) {
System.err.println("Couldn't parse the command line arguments, " +
pe.getMessage());
usage();
}
if (numRows < 1 && inFile == null) usage();
if (numRows > 0 && inFile != null) usage();
if (numMappers > 0 && seed != -1) usage();
if (seed == -1){
seed = System.currentTimeMillis();
}
String remainders[] = opts.getRemainingArgs();
colSpecs = new ColSpec[remainders.length];
for (int i = 0; i < remainders.length; i++) {
colSpecs[i] = new ColSpec(remainders[i]);
}
System.err.println("Using seed " + seed);
rand = new Random(seed);
return 0;
}
private void go() throws IOException {
long t1 = System.currentTimeMillis();
if (numMappers <= 0) {
System.out.println("Generate data in local mode.");
goLocal();
}else{
System.out.println("Generate data in hadoop mode.");
HadoopRunner runner = new HadoopRunner();
runner.goHadoop();
}
long t2 = System.currentTimeMillis();
System.out.println("Job is successful! It took " + (t2-t1)/1000 + " seconds.");
}
public void goLocal() throws IOException {
PrintWriter out = null;
try {
out = new PrintWriter(outputFile);
} catch (FileNotFoundException fnfe) {
System.err.println("Could not find file " + outputFile +
", " + fnfe.getMessage());
return;
} catch (SecurityException se) {
System.err.println("Could not write to file " + outputFile +
", " + se.getMessage());
return;
}
BufferedReader in = null;
if (inFile != null) {
try {
in = new BufferedReader(new FileReader(inFile));
} catch (FileNotFoundException fnfe) {
System.err.println("Unable to find input file " + inFile);
return;
}
}
if (numRows > 0) {
for (int i = 0; i < numRows; i++) {
writeLine(out);
out.println();
}
} else if (in != null) {
String line;
while ((line = in.readLine()) != null) {
out.print(line);
writeLine(out);
out.println();
}
}
out.close();
}
protected void writeLine(PrintWriter out) {
for (int j = 0; j < colSpecs.length; j++) {
if (j != 0) out.print(separator);
// First, decide if it's going to be null
if (rand.nextInt(100) < colSpecs[j].pctNull) {
continue;
}
writeCol(colSpecs[j], out);
}
}
private void writeCol(ColSpec colspec, PrintWriter out) {
switch (colspec.datatype) {
case INT:
out.print(colspec.nextInt());
break;
case LONG:
out.print(colspec.nextLong());
break;
case FLOAT:
out.print(colspec.nextFloat());
break;
case DOUBLE:
out.print(colspec.nextDouble());
break;
case STRING:
out.print(colspec.nextString());
break;
case MAP:
int len = rand.nextInt(20) + 6;
for (int k = 0; k < len; k++) {
if (k != 0) out.print('');
out.print(mapkey[k] + '');
out.print(colspec.gen.randomString());
}
break;
case BAG:
int numElements = rand.nextInt(5) + 5;
for (int i = 0; i < numElements; i++) {
if (i != 0) out.print('');
switch(colspec.contained.datatype) {
case INT: out.print("i"); break;
case LONG: out.print("l"); break;
case FLOAT: out.print("f"); break;
case DOUBLE: out.print("d"); break;
case STRING: out.print("s"); break;
case MAP: out.print("m"); break;
case BAG: out.print("b"); break;
default: throw new RuntimeException("should never be here");
}
writeCol(colspec.contained, out);
}
}
}
private void usage() {
System.err.println("Usage: datagen -rows numrows [options] colspec ...");
System.err.println("\tOptions:");
System.err.println("\t-e -seed seed value for random numbers");
System.err.println("\t-f -file output file, default is stdout");
System.err.println("\t-i -input input file, lines will be read from");
System.err.println("\t\tthe file and additional columns appended.");
System.err.println("\t\tMutually exclusive with -r.");
System.err.println("\t-r -rows number of rows to output");
System.err.println("\t-s -separator character, default is ^A");
System.err.println("\t-m -number of mappers to run concurrently to generate data. " +
"If not specified, DataGenerator runs locally. This option can NOT be used with -e.");
System.err.println();
System.err.print("\tcolspec: columntype:average_size:cardinality:");
System.err.println("distribution_type:percent_null");
System.err.println("\tcolumntype:");
System.err.println("\t\ti = int");
System.err.println("\t\tl = long");
System.err.println("\t\tf = float");
System.err.println("\t\td = double");
System.err.println("\t\ts = string");
System.err.println("\t\tm = map");
System.err.println("\t\tbx = bag of x, where x is a columntype");
System.err.println("\tdistribution_type:");
System.err.println("\t\tu = uniform");
System.err.println("\t\tz = zipf");
throw new RuntimeException();
}
static enum Datatype { INT, LONG, FLOAT, DOUBLE, STRING, MAP, BAG };
static enum DistributionType { UNIFORM, ZIPF };
protected class ColSpec {
String arg;
Datatype datatype;
DistributionType distype;
int avgsz;
int card;
RandomGenerator gen;
int pctNull;
ColSpec contained;
String mapfile;
Map<Integer, Object> map;
public ColSpec(String arg) {
this.arg = arg;
String[] parts = arg.split(":");
if (parts.length != 5 && parts.length != 6) {
System.err.println("Colspec [" + arg + "] format incorrect");
usage();
}
switch (parts[0].charAt(0)) {
case 'i': datatype = Datatype.INT; break;
case 'l': datatype = Datatype.LONG; break;
case 'f': datatype = Datatype.FLOAT; break;
case 'd': datatype = Datatype.DOUBLE; break;
case 's': datatype = Datatype.STRING; break;
case 'm': datatype = Datatype.MAP; break;
case 'b':
datatype = Datatype.BAG;
contained = new ColSpec(arg.substring(1));
return;
default:
System.err.println("Don't know column type " +
parts[0].charAt(0));
usage();
break;
}
avgsz = Integer.valueOf(parts[1]);
card = Integer.valueOf(parts[2]);
switch (parts[3].charAt(0)) {
case 'u':
gen = new UniformRandomGenerator(avgsz, card);
distype = DistributionType.UNIFORM;
break;
case 'z':
gen = new ZipfRandomGenerator(avgsz, card);
distype = DistributionType.ZIPF;
break;
default:
System.err.println("Don't know generator type " +
parts[3].charAt(0));
usage();
break;
}
pctNull = Integer.valueOf(parts[4]);
if (pctNull > 100) {
System.err.println("Percentage null must be between 0-100, "
+ "you gave" + pctNull);
usage();
}
contained = null;
// if config has 6 columns, the last col is the file name
// of the mapping file from random number to field value
if (parts.length == 6) {
mapfile = parts[5];
gen.hasMapFile = true;
}
map = new HashMap<Integer, Object>();
}
public int nextInt() {
return gen.nextInt(map);
}
public long nextLong() {
return gen.nextInt(map);
}
public double nextDouble() {
return gen.nextDouble(map);
}
public float nextFloat() {
return gen.nextFloat(map);
}
public String nextString() {
return gen.nextString(map);
}
}
abstract class RandomGenerator {
protected int avgsz;
protected boolean hasMapFile; // indicating whether a map file from
// random number to the field value is pre-defined
abstract public int nextInt(Map<Integer, Object> map);
abstract public long nextLong(Map<Integer, Object> map);
abstract public float nextFloat(Map<Integer, Object> map);
abstract public double nextDouble(Map<Integer, Object> map);
abstract public String nextString(Map<Integer, Object> map);
public String randomString() {
int var = (int)((double)avgsz * 0.3);
StringBuffer sb = new StringBuffer(avgsz + var);
if (var < 1) var = 1;
int len = rand.nextInt(2 * var) + avgsz - var;
for (int i = 0; i < len; i++) {
int n = rand.nextInt(122 - 65) + 65;
sb.append(Character.toChars(n));
}
return sb.toString();
}
public float randomFloat() {
return rand.nextFloat() * rand.nextInt();
}
public double randomDouble() {
return rand.nextDouble() * rand.nextInt();
}
}
class UniformRandomGenerator extends RandomGenerator {
int card;
public UniformRandomGenerator(int a, int c) {
avgsz = a;
card = c;
}
public int nextInt(Map<Integer, Object> map) {
return rand.nextInt(card);
}
public long nextLong(Map<Integer, Object> map) {
return rand.nextLong() % card;
}
public float nextFloat(Map<Integer, Object> map) {
int seed = rand.nextInt(card);
Float f = (Float)map.get(seed);
if (f == null) {
if (!hasMapFile) {
f = randomFloat();
map.put(seed, f);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return f;
}
public double nextDouble(Map<Integer, Object> map) {
int seed = rand.nextInt(card);
Double d = (Double)map.get(seed);
if (d == null) {
if (!hasMapFile) {
d = randomDouble();
map.put(seed, d);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return d;
}
public String nextString(Map<Integer, Object> map) {
int seed = rand.nextInt(card);
String s = (String)map.get(seed);
if (s == null) {
if (!hasMapFile) {
s = randomString();
map.put(seed, s);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return s;
}
}
class ZipfRandomGenerator extends RandomGenerator {
Zipf z;
public ZipfRandomGenerator(int a, int c) {
avgsz = a;
z = new Zipf(c);
}
// the Zipf library returns a random number [1..cardinality], so we substract by 1
// to get [0..cardinality)
// the randome number returned by zipf library is an integer, but converted into double
private double next() {
return z.nextElement()-1;
}
public int nextInt(Map<Integer, Object> map) {
return (int)next();
}
public long nextLong(Map<Integer, Object> map) {
return (long)next();
}
public float nextFloat(Map<Integer, Object> map) {
int seed = (int)next();
Float d = (Float)map.get(seed);
if (d == null) {
if (!hasMapFile) {
d = randomFloat();
map.put(seed, d);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return d;
}
public double nextDouble(Map<Integer, Object> map) {
int seed = (int)next();
Double d = (Double)map.get(seed);
if (d == null) {
if (!hasMapFile) {
d = randomDouble();
map.put(seed, d);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return d;
}
public String nextString(Map<Integer, Object> map) {
int seed = (int)next();
String s = (String)map.get(seed);
if (s == null) {
if (!hasMapFile) {
s = randomString();
map.put(seed, s);
}else{
throw new IllegalStateException("Number " + seed + " is not found in map file");
}
}
return s;
}
}
// launch hadoop job
class HadoopRunner {
Random r;
FileSystem fs;
Path tmpHome;
public HadoopRunner() {
r = new Random();
}
public void goHadoop() throws IOException {
// Configuration processed by ToolRunner
Configuration conf = getConf();
// Create a JobConf using the processed conf
JobConf job = new JobConf(conf);
fs = FileSystem.get(job);
tmpHome = createTempDir(null);
String config = genMapFiles().toUri().getRawPath();
// set config properties into job conf
job.set("fieldconfig", config);
job.set("separator", String.valueOf((int)separator));
job.setJobName("data-gen");
job.setNumMapTasks(numMappers);
job.setNumReduceTasks(0);
job.setMapperClass(DataGenMapper.class);
job.setJarByClass(DataGenMapper.class);
// if inFile is specified, use it as input
if (inFile != null) {
FileInputFormat.setInputPaths(job, inFile);
job.set("hasinput", "true");
} else {
job.set("hasinput", "false");
Path input = genInputFiles();
FileInputFormat.setInputPaths(job, input);
}
FileOutputFormat.setOutputPath(job, new Path(outputFile));
// Submit the job, then poll for progress until the job is complete
System.out.println("Submit hadoop job...");
RunningJob j = JobClient.runJob(job);
if (!j.isSuccessful()) {
throw new IOException("Job failed");
}
if (fs.exists(tmpHome)) {
fs.delete(tmpHome, true);
}
}
private Path genInputFiles() throws IOException {
long avgRows = numRows/numMappers;
// create a temp directory as mappers input
Path input = createTempDir(tmpHome);
System.out.println("Generating input files into " + input.toString());
long rowsLeft = numRows;
// create one input file per mapper, which contains
// the number of rows
for(int i=0; i<numMappers; i++) {
Object[] tmp = createTempFile(input, false);
PrintWriter pw = new PrintWriter((OutputStream)tmp[1]);
if (i < numMappers-1) {
pw.println(avgRows);
}else{
// last mapper takes all the rows left
pw.println(rowsLeft);
}
pw.close();
rowsLeft -= avgRows;
}
return input;
}
// generate map files for all the fields that need to pre-generate map files
// return a config file which contains config info for each field, including
// the path to their map file
private Path genMapFiles() throws IOException {
Object[] tmp = createTempFile(tmpHome, false);
System.out.println("Generating column config file in " + tmp[0].toString());
PrintWriter pw = new PrintWriter((OutputStream)tmp[1]);
for(int i=0; i<colSpecs.length; i++) {
DataGenerator.Datatype datatype = colSpecs[i].datatype;
pw.print(colSpecs[i].arg);
if ( datatype == DataGenerator.Datatype.FLOAT || datatype == DataGenerator.Datatype.DOUBLE ||
datatype == DataGenerator.Datatype.STRING) {
Path p = genMapFile(colSpecs[i]);
pw.print(':');
pw.print(p.toUri().getRawPath());
}
pw.println();
}
pw.close();
return (Path)tmp[0];
}
// genereate a map file between random number to field value
// return the path of the map file
private Path genMapFile(DataGenerator.ColSpec col) throws IOException {
int card = col.card;
Object[] tmp = createTempFile(tmpHome, false);
System.out.println("Generating mapping file for column " + col.arg + " into " + tmp[0].toString());
PrintWriter pw = new PrintWriter((OutputStream)tmp[1]);
HashSet<Object> hash = new HashSet<Object>(card);
for(int i=0; i<card; i++) {
pw.print(i);
pw.print("\t");
Object next = null;
do {
if (col.datatype == DataGenerator.Datatype.DOUBLE) {
next = col.gen.randomDouble();
}else if (col.datatype == DataGenerator.Datatype.FLOAT) {
next = col.gen.randomFloat();
}else if (col.datatype == DataGenerator.Datatype.STRING) {
next = col.gen.randomString();
}
}while(hash.contains(next));
hash.add(next);
pw.println(next);
if ( (i>0 && i%300000 == 0) || i == card-1 ) {
System.out.println("processed " + i*100/card + "%." );
pw.flush();
}
}
pw.close();
return (Path)tmp[0];
}
private Path createTempDir(Path parentDir) throws IOException {
Object[] obj = createTempFile(parentDir, true);
return (Path)obj[0];
}
private Object[] createTempFile(Path parentDir, boolean isDir) throws IOException {
Path tmp_home = parentDir;
if (tmp_home == null) {
tmp_home = new Path(fs.getHomeDirectory(), "tmp");
}
if (!fs.exists(tmp_home)) {
fs.mkdirs(tmp_home);
}
int id = r.nextInt();
Path f = new Path(tmp_home, "tmp" + id);
while (fs.exists(f)) {
id = r.nextInt();
f = new Path(tmp_home, "tmp" + id);
}
// return a 2-element array. first element is PATH,
// second element is OutputStream
Object[] r = new Object[2];
r[0] = f;
if (!isDir) {
r[1] = fs.create(f);
}else{
fs.mkdirs(f);
}
return r;
}
}
public static class DataGenMapper extends MapReduceBase implements Mapper<LongWritable, Text, String, String> {
private JobConf jobConf;
private DataGenerator dg;
private boolean hasInput;
public void configure(JobConf jobconf) {
this.jobConf = jobconf;
int id = Integer.parseInt(jobconf.get(MRConfiguration.TASK_PARTITION));
long time = System.currentTimeMillis() - id*3600*24*1000;
dg = new DataGenerator( ((time-id*3600*24*1000) | (id << 48)));
dg.separator = (char)Integer.parseInt(jobConf.get("separator"));
if (jobConf.get("hasinput").equals("true")) {
hasInput = true;
}
String config = jobConf.get("fieldconfig");
try {
FileSystem fs = FileSystem.get(jobconf);
// load in config file for each column
BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(new Path(config))));
String line = null;
List<DataGenerator.ColSpec> cols = new ArrayList<DataGenerator.ColSpec>();
while((line = reader.readLine()) != null) {
cols.add(dg.new ColSpec(line));
}
reader.close();
dg.colSpecs = cols.toArray(new DataGenerator.ColSpec[0]);
// load in mapping files
for(int i=0; i<dg.colSpecs.length; i++) {
DataGenerator.ColSpec col = dg.colSpecs[i];
if (col.mapfile != null) {
reader = new BufferedReader(new InputStreamReader(fs.open(new Path(col.mapfile))));
Map<Integer, Object> map = dg.colSpecs[i].map;
while((line = reader.readLine()) != null) {
String[] fields = line.split("\t");
int key = Integer.parseInt(fields[0]);
if (col.datatype == DataGenerator.Datatype.DOUBLE) {
map.put(key, Double.parseDouble(fields[1]));
}else if (col.datatype == DataGenerator.Datatype.FLOAT) {
map.put(key, Float.parseFloat(fields[1]));
}else {
map.put(key, fields[1]);
}
}
reader.close();
}
}
}catch(IOException e) {
throw new RuntimeException("Failed to load config file. " + e);
}
}
public void map(LongWritable key, Text value, OutputCollector<String, String> output, Reporter reporter) throws IOException {
int intialsz = dg.colSpecs.length * 50;
if (!hasInput) {
long numRows = Long.parseLong(value.toString().trim());
dg.numRows = numRows;
for (int i = 0; i < numRows; i++) {
StringWriter str = new StringWriter(intialsz);
PrintWriter pw = new PrintWriter(str);
dg.writeLine(pw);
output.collect(null, str.toString());
if ((i+1) % 10000 == 0) {
reporter.progress();
reporter.setStatus("" + (i+1) + " tuples generated.");
}
}
} else {
StringWriter str = new StringWriter(intialsz);
PrintWriter pw = new PrintWriter(str);
pw.write(value.toString());
dg.writeLine(pw);
output.collect(null, str.toString());
}
}
}
}