2

我使用 Datastax Enterprise 4.5。我希望我的配置是正确的，就像在 datastax 网站上解释的那样。我可以使用 Windows 服务写入 Cassandra DB，这可行，但我无法使用 where 函数使用 Spark 进行查询。

我使用“./dse cassandra -k -t”（在 /bin 文件夹中）启动 Cassandra 节点（只有一个用于测试目的），因此 hadoop 和 spark 都在运行。我可以毫无问题地写入 Cassandra。

因此，当“where”不是 RowKey 时，您不能在 Cassandra 查询中使用“where”子句。所以我需要使用 Spark/Shark。我可以使用鲨鱼 (./dse shark) 启动和使用我需要的所有查询，但我需要用 Scala 或 Java 编写独立程序。

所以我尝试了这个链接：https ://github.com/datastax/spark-cassandra-connector

我可以查询一个简单的语句，例如：

val conf = new SparkConf(true)
  .set("spark.cassandra.connection.host", "MY_IP")
  .setMaster("spark://MY_IP:7077")
  .setAppName("SparkTest")

// Connect to the Spark cluster:
lazy val sc = new SparkContext(conf)

val rdd = sc.cassandraTable("keyspace", "tablename")
println(rdd.first)

这很好用，但如果我要求更多行或计数：

println(rdd.count)
rdd.toArray.foreach(println)

然后我得到这个异常：

Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: All masters are unresponsive! Giving up.
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1049)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1033)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1031)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1031)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:635)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:635)
at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1234)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

当我在 Java 中尝试这个时，我遇到了同样的问题。有谁知道这个问题？我不知道数据库配置是否正确，或者 scala/Javaprogram 是否正常工作。也许某些端口 a 被阻止，但 7077 和 4040 是开放的。

旁注：如果我在 Cassandra DB 上启动 spark，我可以执行以下查询：

sc.cassandraTable("test","words").select("word").toArray.foreach(println)

但是，如果我使用“where”子句，例如：

sc.cassandraTable("test","words").select("word").where("word = ?","foo").toArray.foreach(println)

我得到这个例外：

java.io.IOException: Exception during query execution: SELECT "word" FROM "test"."words" WHERE token("word") > 0 AND word = ? ALLOW FILTERING

你有什么想法吗？我以为我可以在 spark 中使用 where 子句？

谢谢！

4

2 回答 2

2 
All masters are unresponsive!

表示您尝试连接的 IP 实际上不受 spark 的约束。所以这基本上是一个网络配置错误。扫描以查看哪些接口正在侦听 7077，并确保您连接到正确的接口。

至于第二个问题，where运算符暗示您将对该子句进行谓词下推。目前，您无法使用主键执行此操作。如果您想where使用单个主键，您可以执行此filter操作，但您不会看到出色的性能，因为这将执行整个表扫描。

于 2014-09-01T18:51:30.643 回答
0

到目前为止，这是我的解决方案。这不是我所有问题的答案，但它对我有用，我想分享给你。

我使用 hive jdbc 驱动程序通过 Java 访问 SharkServer。这个怎么运作：

启动鲨鱼服务器：bin/dse shark --service sharkserver -p <port>

Maven的依赖：

<dependency>
    <groupId>org.apache.hive</groupId>
    <artifactId>hive-jdbc</artifactId>
    <version>0.13.1</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-core</artifactId>
    <version>0.20.2</version>
</dependency>

Java 代码：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public class HiveJdbcClient {
  private static String driverName = "org.apache.hadoop.hive.jdbc.HiveDriver";

  public static void main(String[] args) throws SQLException {
    try {
      Class.forName(driverName);
    } catch (ClassNotFoundException e) {
      e.printStackTrace();
      System.exit(1);
    }
    Connection con = DriverManager.getConnection("jdbc:hive://YOUR_IP:YOUR_PORT/default", "", "");
    Statement stmt = con.createStatement();
    String sql;
    ResultSet res;



    sql = "SELECT * FROM keyspace.colFam WHERE name = 'John'";
    res = stmt.executeQuery(sql);
    while (res.next()) {
        System.out.println(res.getString("name"));
   }
 }
}
于 2014-09-03T07:19:57.860 回答