MapReduce with Vertica
Hi All,
I am trying to write into vertica table from Mapreduce App
Am following this link
package tut.Vertica;
import java.io.IOException;
import java.sql.SQLException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import com.vertica.hadoop.VerticaOutputFormat;
import com.vertica.hadoop.VerticaRecord;
public class VerticaWrite extends Configured implements Tool
{
public static class VerticaMapper extends Mapper<Object, Text, Text, IntWritable>
{
Text word = new Text();
IntWritable one = new IntWritable(1);
@Override
protected void map(Object key, Text value, Context context) throws IOException, InterruptedException
{
StringTokenizer tokens = new StringTokenizer(value.toString());
while(tokens.hasMoreTokens())
{
word.set(tokens.nextToken());
context.write(word, one);
}
}
}
public static class VerticaReducer extends Reducer<Text, Text, WritableComparable, VerticaRecord>
{
VerticaRecord record = null;
@Override
protected void setup(org.apache.hadoop.mapreduce.Reducer.Context context)
throws IOException, InterruptedException
{
try {
record = new VerticaRecord(context.getConfiguration());
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (SQLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException
{
int sum =0 ;
for(IntWritable i : values)
{
sum = sum + i.get();
}
if (record == null) {
throw new IOException("No output record found");
}
record.set(0, key.toString());
record.set(1,sum);
context.write(new Text("someSchema.SomeTarget"), record);
// write context object into vertica
}
}
public static void main(String[] args) throws Exception
{
int res = ToolRunner.run(new Configuration(), new VerticaWrite(), args);
System.exit(res);
}
public int run(String[] args) throws Exception
{
Configuration conf = new Configuration();
conf.set("mapred.vertica.hostnames", "somehostname.com");
conf.set("mapred.vertica.port","5433");
conf.set("mapred.vertica.username","someuser");
conf.set("mapred.vertica.password","somepassword");
conf.set("mapred.vertica.database","somedb");
Job job = Job.getInstance(conf, "This is to read Vertica table");
job.setInputFormatClass(TextInputFormat.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(VerticaRecord.class);
job.setMapperClass(VerticaMapper.class);
job.setReducerClass(VerticaReducer.class);
// setting output format class here ie to vertica stuff
FileInputFormat.addInputPath(job, new Path(args[0])) ;
job.setOutputFormatClass(VerticaOutputFormat.class);
VerticaOutputFormat.setOutput(job, "someSchema.SomeTarget",true, "v varchar","c int");
job.waitForCompletion(true);
return 0;
}
}
Used the below line to run the hadoop command
hadoop jar vwrite.jar tut.Vertica.VerticaWrite -libjars /usr/hdp/2.3.4.0-3485/hadoop/lib/hadoop-vertica.jar /hp/caferacer
INFO client.RMProxy: Connecting to ResourceManager at xxxxxxx.xxxx.xxxxxx.net/xx.xx.xx.xx:8050
Exception in thread "main" java.lang.RuntimeException: java.sql.SQLSyntaxErrorException: [Vertica][VJDBC](4368) ROLLBACK: Permission denied for database [somedb]
at com.vertica.hadoop.VerticaOutputFormat.checkOutputSpecs(VerticaOutputFormat.java:115)
at com.vertica.hadoop.VerticaOutputFormat.checkOutputSpecs(VerticaOutputFormat.java:74)