java - Cassandra中的Hadoop身份验证不起作用

Question

当我尝试使用仅 map 的hadoop作业将数据插入cassandra时出现以下错误。我目前正在使用cassandra 2.0.6 jar

我的驱动程序类中用于身份验证的代码段以及下面的身份验证错误

@Override
public int run(String[] arg0) throws Exception {
    Configuration conf = new Configuration();
    ConfigHelper.setOutputColumnFamily(conf, "mykeyspace", "my_records");
    ConfigHelper.setOutputRpcPort(conf, "9160");
    ConfigHelper.setOutputKeyspaceUserName(conf, "myusername");
    ConfigHelper.setOutputKeyspacePassword(conf, "mypassword");
    conf.set("mapreduce.output.bulkoutputformat.buffersize", "64");
    conf.setInt("mapreduce.linerecordreader.maxlength", 10);
    CqlConfigHelper.setOutputCql(conf, query);
    ConfigHelper.setOutputInitialAddress(conf, "10.211.213.59");
    ConfigHelper.setOutputPartitioner(conf, "Murmur3Partitioner");      

    Job job = new Job(conf);
    job.setJobName("Cassandra CqlBulk Loader");       
    job.setOutputFormatClass(CqlOutputFormat.class);
    job.setInputFormatClass(MyFileInputFormat.class);
    job.setMapOutputKeyClass(Map.class);
    job.setMapOutputValueClass(List.class);
    job.setMapperClass(MyCqlMapper.class);
    job.setJar("cqlbulkloader-hadoop.jar");
    job.setNumReduceTasks(0);
    TextInputFormat.addInputPaths(job, "/input/ip_file.txt");
    return job.waitForCompletion(true) ? 0 :1;
}

错误堆栈跟踪

Error: java.lang.RuntimeException: InvalidRequestException(why:You have not logged in)
    at org.apache.cassandra.hadoop.cql3.CqlRecordWriter.<init>(CqlRecordWriter.java:121)
    at org.apache.cassandra.hadoop.cql3.CqlRecordWriter.<init>(CqlRecordWriter.java:88)
    at org.apache.cassandra.hadoop.cql3.CqlOutputFormat.getRecordWriter(CqlOutputFormat.java:74)
    at org.apache.cassandra.hadoop.cql3.CqlOutputFormat.getRecordWriter(CqlOutputFormat.java:55)
    at org.apache.hadoop.mapred.MapTask$NewDirectOutputCollector.<init>(MapTask.java:623)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:743)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:339)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:162)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)
    at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:157)
Caused by: InvalidRequestException(why:You have not logged in)
    at org.apache.cassandra.thrift.Cassandra$execute_cql3_query_result$execute_cql3_query_resultStandardScheme.read(Cassandra.java:49032)
    at org.apache.cassandra.thrift.Cassandra$execute_cql3_query_result$execute_cql3_query_resultStandardScheme.read(Cassandra.java:49009)
    at org.apache.cassandra.thrift.Cassandra$execute_cql3_query_result.read(Cassandra.java:48924)
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:78)
    at org.apache.cassandra.thrift.Cassandra$Client.recv_execute_cql3_query(Cassandra.java:1693)
    at org.apache.cassandra.thrift.Cassandra$Client.execute_cql3_query(Cassandra.java:1678)
    at org.apache.cassandra.hadoop.cql3.CqlRecordWriter.retrievePartitionKeyValidator(CqlRecordWriter.java:335)
    at org.apache.cassandra.hadoop.cql3.CqlRecordWriter.<init>(CqlRecordWriter.java:106)

有任何解决方法？

还有一件事

为什么使用hadoop在CqlRecordWriter cassandra的写入中不支持插入？
Link to code

Answer 1

贴剂的作用就像面霜。 https://github.com/apache/cassandra/pull/27.patch
补丁程序针对Cassandra 2.0.10

将以下代码添加到CqlRecordWriter

         try
         {
             Cassandra.Client client = ConfigHelper.getClientFromOutputAddressList(conf);
+
+            if (ConfigHelper.getOutputKeyspaceUserName(conf) != null)
+            {
+                Map<String, String> credentials = new HashMap<String, String>();
+                credentials.put(IAuthenticator.USERNAME_KEY, ConfigHelper.getOutputKeyspaceUserName(conf));
+                credentials.put(IAuthenticator.PASSWORD_KEY, ConfigHelper.getOutputKeyspacePassword(conf));
+                AuthenticationRequest authRequest = new AuthenticationRequest(credentials);
+                client.login(authRequest);
+            }
+
             retrievePartitionKeyValidator(client);
             String cqlQuery = CqlConfigHelper.getOutputCql(conf).trim();
             if (cqlQuery.toLowerCase().startsWith("insert"))