问题标签 [spark-cassandra-connector]

我想将 Kafka + Cassandra 连接到 Spark 1.5.1。

库的版本：

app的初始化和使用：

像这样在 Cassandra 中创建模式：

同样在准备好的时候jar，制定一些策略：

我认为这个问题与

捆绑使用MergeStrategy.last。

有任何想法吗？

有异常：

我正在并行化 spark 中的查询列表，并且我的 RDD 被映射到对象集合中，其中对象类型会根据每个查询的数据源属性而不同。我有一个 String -> 函数的映射，其中每个函数都为我提供了要保存到 cassandra 中的对象集合。例如：

其中 processTrendsResponse 被定义为返回 Seq[Trends] 和 processYahooResponse 返回 Seq[Yahoo] 的函数，其中 Trends 和 Yahoo 定义被定义为案例类：

为了适应 Trends 和 Yahoo 类型，processingMethods Map 被定义为 Map[String, (String) => Seq[Any]]。但是当我在 spark 中执行 saveToCassandra 操作时，我遇到了一个例外 -

提前致谢

我是 Spark 和 Cassandra 的新手。我面临一个主要的性能问题。我在 Spark 中每 5 秒从 Kafka 流式传输数据，然后使用 JRI 对 R 语言中的数据执行分析，最后将数据保存到 Cassandra 各自的列族。将数据保存到 Cassandra 的持续时间（以毫秒为单位）随着输入请求数量的增加而迅速增加 [每个请求为 200KB]。

火花代码：

我使用 Spark SQL 从 Cassandra 数据库中检索数据：

之后我进行了一些过滤，我想将这些数据保存到另一个 Cassandra 表中，如下所示：

DataFrame保存到新表时如何添加这些附加属性？另外，使用此示例对 Cassandra 长行进行分片的最佳做法是什么？我希望在 4k-6k 记录中DataFrame，因此必须对长行进行分片，但我不确定计算记录然后更改sharder一定数量的项目是否是 Spark 或 Cassandra 中的最佳实践。

我正在尝试让 Cassandra 2.2.3（在 localhost 上运行）与 Spark 1.5.1 和最新版本的 spark-cassandra-connector (1.5.0-M2) 一起使用。

这是我正在使用的基本代码段。键空间和表已经创建。

sbt 程序集编译时没有任何问题，但在提交应用程序时出现此错误：

我创建了一个简单的 Java 应用程序，它使用 Apache Spark 从 Cassandra 检索数据，对其进行一些转换并将其保存在另一个 Cassandra 表中。

我正在使用以独立集群模式配置的 Apache Spark 1.4.1，在我的机器上只有一个主从模式。

如您所见，我cache是customersWhoHaventOrderedTheProductDataFrame，之后我执行 acount并调用toJavaRDD.

根据我的计算，这些动作应该只执行一次。但是，当我进入当前工作的 Spark UI 时，我会看到以下阶段：

如您所见，每个动作都执行了两次。

难道我做错了什么？有没有我错过的设置？

任何想法都非常感谢。

编辑：

我打电话后System.out.println(storeCustomerReport.toJavaRDD().toDebugString());

这是调试字符串：

编辑2：

因此，经过一些研究并结合试验和错误，我设法优化了这项工作。

我创建了一个 RDD，customersWhoHaventOrderedTheProduct并在调用操作之前将其缓存count()。（我将缓存从 移动DataFrame到RDD）。

之后，我使用它RDD来创建storeCustomerReport DataFrame.

现在阶段看起来像这样：

如您所见，这两个count现在cache都消失了，但仍然有两个“javaRDD”操作。我不知道它们来自哪里，因为我toJavaRDD在代码中只调用了一次。

Spark 环境的新手（对 Maven 来说也很新），所以我正在努力解决如何正确发送我需要的依赖项。

看起来 Spark 1.5.1 有一个它尝试使用的 guava-14.0.1 依赖项，并且 isPrimitive 是在 15+ 中添加的。确保我的 uber-jar 获胜的正确方法是什么？我已经尝试spark.executor.extraClassPath在我的 spark-defaults.conf 中无济于事。

复制到这个[问题]：Spark 1.5.1 + Scala 2.10 + Kafka + Cassandra = Java.lang.NoSuchMethodError: but for Maven 本质上（还没有代表发表评论）

将我的依赖项剥离到这个：

使用以下所有依赖项为我的 JAR 着色：

这是我跑步时的惊人爆炸

./spark-submit --master local --class <my main class> <my shaded jar>

我在 Spark 中有以下代码：

在哪里

在哪里：

saveToCassandra期望对象集合并Seq[Any]用作返回类型以包含两者Seq[A]并以异常Seq[B]中断-不是一个术语。这种行为的原因可能是什么？saveToCassandrascala.ScalaReflectionException: <none>

我想读取文件数据并检查 Cassandra 中是否存在文件行数据，如果存在则需要合并，否则将新插入到 C*。文件数据只包含名称，json格式的地址，在Cassandra学生表中以UUID为主键，名称上有二级索引

将数据合并到 cassandra 后，我想将新的 UUID 或现有的 UUID 发送到 KAfka。

当我在 mesos 集群上的本地或单台机器上运行时（保持线 sparkConf.setMaster("local[4]");) 这个程序可以工作但是当我提交给带有 4 个从属的 mesos master 时（注释行 //sparkConf.setMaster(" local[4]"); 在集群上）在 javastreaming 上下文中从 Cassandra 选择数据时存在空指针

我将流上下文设为静态，因为它在文件 dstream 的映射转换中被访问时引发了序列化异常。

这种方法有什么问题吗？是因为我正在尝试使用 DStream 地图转换构建 Cassandra RDD，这会导致问题吗？

异常堆栈跟踪::

从 Cassandra 表加载数据时，spark 分区表示具有相同分区键的所有行。但是，当我使用相同的分区键在 spark 中创建数据并使用 .repartitionByCassandraReplica(..) 方法对新的 RDD 重新分区时，它最终会出现在不同的 spark 分区中吗？如何使用 Spark-Cassandra 连接器定义的分区方案在 Spark 中实现一致的分区？

下载我测试的 CQL 和 Spark 作业代码的链接

• .CQL 与键空间和表模式。
• Spark 作业和其他类。

版本和其他信息

• 火花：1.3
• 卡桑德拉：2.1
• 连接器：1.3.1
• Spark 节点 (5) 和 Cass* 集群节点 (4) 在不同的数据中心运行

代码提取。使用上面的链接下载代码以获取更多详细信息

第 1 步：将数据加载到 8 个 spark 分区中

第 2 步：将数据重新分区为 8 个分区

第 3 步：打印两个 rdds 的分区 id 和值

第 4 步：在分区 1 上打印的结果快照。两个 Rdd 不同，但期望相同

加载 Rdd 值