Hi, I have a Java code that read a file and insert the data in a hbase table. Without Kerberos it's working well.
I tried my code on a cluster with isilon and kerberos enable. The part of the code that read the file and count the number of lines work well but once it come the code where he use the function saveAsNewAPIHadoopDataset, nothing happen. The appliction seams to be stuck, no log are print in the shell or in any executor. Here you have the code use: public static void ingestProcodacFolderKerberosDev(String file, int numP, String table) {
try {
String zookeeper = "zk0001.bc,zk0002.bc,zk0003.bc";
Configuration confFS = new Configuration();
confFS.addResource("/etc/hadoop/conf/core-site.xml");
confFS.addResource("/etc/hadoop/conf/hdfs-site.xml");
FileSystem dfs = FileSystem.get(confFS);
SparkConf conf = new SparkConf().setAppName("TEST - ingest Procodac Dev Ker");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> procodac = sc.textFile(file, numP);
Configuration config = null;
try {
config = HBaseConfiguration.create();
config.set("hbase.zookeeper.quorum", zookeeper);
config.set("zookeeper.znode.parent", "/hbase-secure");
config.addResource("/user/hdp/current/hbase-client/config/core-site.xml");
config.addResource("/user/hdp/current/hbase-client/config/hbase-site.xml");
config.addResource("/user/hdp/current/hbase-client/config/hdfs-site.xml");
} catch (Exception ce) {
ce.printStackTrace();
}
JavaRDD<String> procodacFiltre = procodac.filter(new Function<String, Boolean>() {
public Boolean call(String s) {
return (s.split("\t").length >= 4);
}
});
System.out.println("!!!!!!!!!!!! " + procodacFiltre.count());
config.set(TableOutputFormat.OUTPUT_TABLE, table);
Job newAPIJobConfiguration1 = Job.getInstance(config);
newAPIJobConfiguration1.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, table);
newAPIJobConfiguration1.setOutputFormatClass(org.apache.hadoop.hbase.mapreduce.TableOutputFormat.class);
JavaPairRDD<ImmutableBytesWritable, Put> hbasePuts = procodacFiltre.mapToPair(
new PairFunction<String, ImmutableBytesWritable, Put>() {
@Override
public Tuple2<ImmutableBytesWritable, Put> call(String t) throws Exception {
String[] elements = t.split("\t");
String id;
String colFam;
id = elements[0] + "_" + elements[2];
if (t.contains("DOWNstream")) {
colFam = "downstream";
} else {
colFam = "upstream";
}
String fields[] = new String[4];
fields[0] = "phone";
fields[1] = "port";
fields[2] = "date";
fields[3] = "type";
Put put = new Put(Bytes.toBytes(id));
for (int g = 0; g < elements.length; g++) {
if (fields[g] != null) {
put.add(Bytes.toBytes(colFam), Bytes.toBytes(fields[g]), Bytes.toBytes(elements[g]));
}
}
return new Tuple2<>(new ImmutableBytesWritable(id.getBytes()), put);
}
});
// save to HBase
hbasePuts.saveAsNewAPIHadoopDataset(newAPIJobConfiguration1.getConfiguration());
} catch (IOException ex) {
Logger.getLogger(procodac.class.getName()).log(Level.SEVERE, null, ex);
}
} Before starting the application I'm doing a kinit command, then the spark-submit. I know htat the kerberos authentification is working because the spark jobis able to read the file on hdfs and to count hte number of line as you can see in the following screenshot (second line:" !!!!!!!!! 4247031") At the end of the screenshot you can see that it start several task, but nothing happen anywhere. For info, I'm using HDP2.5 with Spark 1.6.2 Does any one have an idea how to solve that? Thanks in advance for your help 🙂 Michel
... View more
