I am trying to write a hive udf which connects to hbase table. But, the program is failing to access due to security exception and throwing below: javax.security.auth.login.LoginException: Unable to obtain password from user at Below is the code i am trying ... config = HBaseConfiguration.create();         config.set("hadoop.security.authentication", "Kerberos");         config.set("hbase.security.authentication", "kerberos");         config.addResource("src/main/resources/hbase-site.xml");         // Point to the krb5.conf file.         System.setProperty("java.security.krb5.conf", "src/main/resources/krb5.conf");         System.setProperty("sun.security.krb5.debug", "true");         UserGroupInformation.setConfiguration(config);         UserGroupInformation.loginUserFromKeytab("pricipal", "keytab");         connection = ConnectionFactory.createConnection(config); can someone please share ur thoughts and help me fix the issue .... ... View more

I have written a bare minimum storm topology, to read data from kafka and directly dump into hdfs! Storm topology doesn't show any exceptions on storm ui,But it is not downloading any messages also from Kafka..! cannot debug also! [I am not sure where the logs will be saved. etc. any help pls ] I have set storm kafka spout to setting "OffsetRequest.EarliestTime();" to ensure download messages from the start...! I know there are messages in the topic, but my storm topology is not downloading anything! Can someone pls share your thoughts to help me fix this ... Also, few details, i have built a storm-kafka module seperately, as i was informed, that storm in hdp2.5 will only support hwx kafka and kerberos secured one... ! But our kafka environment is not both! hence took apache source code and compiled it! Please help me fix this issue. Thanks a tonnn!!! Attached the pom.xml's i used to build this. Master pomfile: masterpom.xml pom file used to build storm-kafka module: storm-kafkapom.xml pom file used to actual topology: storm-topology-actual-sourcepom.xml Find below the source code: import kafka.api.OffsetRequest; import org.apache.storm.Config; import org.apache.storm.StormSubmitter; import org.apache.storm.generated.AlreadyAliveException; import org.apache.storm.generated.AuthorizationException; import org.apache.storm.generated.InvalidTopologyException; import org.apache.storm.generated.StormTopology; import org.apache.storm.hdfs.bolt.HdfsBolt; import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat; import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat; import org.apache.storm.hdfs.bolt.format.FileNameFormat; import org.apache.storm.hdfs.bolt.format.RecordFormat; import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy; import org.apache.storm.hdfs.bolt.rotation.FileSizeRotationPolicy; import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy; import org.apache.storm.hdfs.bolt.sync.SyncPolicy; import org.apache.storm.kafka.*; import org.apache.storm.spout.SchemeAsMultiScheme; import org.apache.storm.topology.TopologyBuilder; import java.io.IOException; import java.util.HashMap; import java.util.Map; import java.util.UUID; /** * Created by z013sqm on 3/17/17. */ public class KafkaToHdp { public static void main(String[]args) throws IOException, InvalidTopologyException, AlreadyAliveException, AuthorizationException { //Building Config Maps for external system integrations String hdfsConfigKey = "hdfs.config"; Map<String, Object> hdfsConfigMap = new HashMap<String, Object>(); hdfsConfigMap.put("hdfs.keytab.file", "keytab path"); hdfsConfigMap.put("hdfs.kerberos.principal", "keytab princ"); Config configured = new Config(); configured.setDebug(true); configured.put(hdfsConfigKey, hdfsConfigMap); configured.setNumWorkers(1); configured.setMaxSpoutPending(1000); configured.setNumAckers(1); configured.setMessageTimeoutSecs(600); BrokerHosts hosts = new ZkHosts("xxxx"); SpoutConfig spoutConfig = new SpoutConfig(hosts, "xxxx", "/storm" + "/xxxx", UUID.randomUUID().toString()); spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); spoutConfig.startOffsetTime = OffsetRequest.EarliestTime(); KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig); //Settings to setup HDFS Write Bolt FileNameFormat fileNameWithPath = new DefaultFileNameFormat().withPath("xxxx"); RecordFormat recFormat = new DelimitedRecordFormat().withFieldDelimiter("^^^^^"); SyncPolicy syncPolicy = new CountSyncPolicy(5); FileRotationPolicy fileRotatePolicy = new FileSizeRotationPolicy(256.0f, FileSizeRotationPolicy.Units.MB); HdfsBolt hdfsbolt = new HdfsBolt() .withFsUrl("xxxxx") .withRecordFormat(recFormat) .withFileNameFormat(fileNameWithPath) .withRotationPolicy(fileRotatePolicy) .withSyncPolicy(syncPolicy) .withConfigKey(hdfsConfigKey); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("kafka-spout", kafkaSpout, 2); builder.setBolt("write-to-hdfs", hdfsbolt).shuffleGrouping("kafka-spout"); StormTopology topology = builder.createTopology(); StormSubmitter.submitTopology("testing-storm-hdp-2p5_test", configured, topology); } } ... View more

Hi I wrote an MR application that does a lookup of HBase table without extending TableMapper/TableReducer classes!! The reason why i did not extend TableMapper/TableReducer classes is, HBase Table is not a source/destination for me, rather, i want to do a lookup on some table based on the content of my raw input !! The Hadoop cluster i am executing is a secure cluster... which is what causing my application to not execute and throw some errors regarding to security ! Below is the sample code i wrote: public class MyMapper extends Mapper<LongWritable, Text, Text, Text> { private Logger LOG = LoggerFactory.getLogger(MyMapper.class); public void setup(Context context) throws IOException, InterruptedException { System.out.println("Entered Setup method in Mapper."); LOG.info("Entered Setup method in Mapper."); Configuration config = HBaseConfiguration.create(); Connection connection = ConnectionFactory.createConnection(config); Table table = connection.getTable(TableName.valueOf(<HBASE_Table>)); Get get = new Get(Bytes.toBytes(<row_key>)); get.addColumn(Bytes.toBytes(<col_family_name>), Bytes.toBytes(<qualifier_name>)); Result result = table.get(get); byte[] latitude = result.getValue(Bytes.toBytes(<col_family_name>), Bytes.toBytes(<qualifier_name>)); System.out.println(Bytes.toString(<qualifier_name>).toString()); table.close(); connection.close(); System.out.println("completed Setup method in Mapper."); LOG.info("completed Setup method in Mapper."); } public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //// dummy code goes here } } Please find below the exception i am facing: WARN [main-SendThread(localhost.localdomain:2181)] org.apache.zookeeper.ClientCnxn: Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect java.net.ConnectException: Connection refused at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1125) Can someone pls help me fix the issue ?? ... View more

Hi, I have an Apache Apex application which runs on my Hadoop environment. I have no problem with the application except that, it is failing after 7days. And, i realized that it is because of the cluster level setting for any application. Is there any way, i can renew the delegation token perodically at some interval to ensure job runs continously without failing!! I could find any resources online for on how to renew a hdfs delegation tokens!! Can someone please share your knowledge ? ... View more

Hi I know, storm doesn't guarantee total ordering gurantee for kafka topics, but see in many documents, storm guarantees consumption/processing the messages maintaining the order at partition level. I am looking for a sample storm topology, that consumes/processes the messages of a kafka topic maintaining the order of messages at a kafka partition level.. NOT Total Order!! ONLY partition level ordering guarantee. please share if you know any sample application. Thanks a lot!! ... View more

Hi i had a storm topology that reads records from kafka, extracts timestamp present in the record, and does a lookup on hbase table, apply business logic, and then updates the hbase table with latest values in the current record!! I have written a custom hbase bolt extending BaseRichBolt, where, the code, does a lookup on the hbase table and apply some business logic on the message that has been read from kafka, and then updates the hbase table with latest data! The problem i am seeing is, some times, the bolt is receiving/processing the records in a jumbled order, due to which my application is thinking that a particular record is already processed, and ignoring the record!!! Application is not processing a serious amount of records due to this!! For Example: suppose there are two records that are read from kafka, one record belongs to 10th hour and second records belongs to 11th hour... My custom HBase bolt, processing the 11th hour record first... then reading/processing the 10th hour record later!! Because, 11th hour record is processed first, application is assuming 10th record is already processed and ignoring the 10th hour record from processing!! Can someone pls help me understand, why my custom hbase bolt is not processing the records in order it receive ? should i have to mention any additional properties to ensure, the bolt processes the records in the order it receives ? what are possible alternatives i can try to fix this ? FYI, i am using field grouping for hbase bolt, thru which i want to ensure, all the records of a particular user goes into same task!! Nevertheless to mention, thinking field grouping might causing the issue, reduces the no.of tasks for my custom hbase bolt to 1 task, still the same issue!! Wondering why hbase bolt is not reading/processing records in the order it receives !!! Please someone help me with your thoughts!! Thanks a lot. ... View more

Hi would like to know, if its possible to retrieve kafka offset also along with the message from KafkaSpout, when using "storm-kafka" integration ?? If its possbile, can someone please help me with some directions ? ... View more

Hi, can someone help me understand, what happens when i append data to a hdfs file but doesn't sync/flush the contents ?? Will the data appended reflects into the hdfs file after sometime ? or the data won't reflect untill we flush ? My problem is, Using storm-hdfs integration, i am writing to a file in hdfs, and configured application to sync for every 1000 records! Lets assume, storm wrote some 750 records to hdfs file... and later i killed my topology.... and therefore no records will be added to the hdfs file now, and never become 1000 to sync/flush!! Now what would happen to those 750records!! Will they ever reflect into my hdfs file ??? OR i should consider them as "Records are LOST!!" ??? Thanks a lot in advance for your thoughts!! ... View more