在java中向cassandra表写入Dataframe

cwxwcias 于 2021-06-10 发布在 Cassandra

关注(0)|答案(1)|浏览(361)

找不到我需要的东西。scala和python中的大量代码。以下是我所拥有的：

import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
public class CassandraWriter {
    private transient Logger logger = Logger.getLogger(CassandraWriter.class);
    private Dataset<Row> hdfsDF;
    public CassandraWriter(Dataset<Row> dataFrame) {
        hdfsDF = dataFrame;
    }
    public void writeToCassandra(String tableName, String keyspace) {
        logger.info("Writing DataFrame to table: " + tableName);
        hdfsDF.write().format("org.apache.spark.sql.cassandra").mode("overwrite")
                .option("table",tableName)
                .option("keyspace",keyspace)
                .save();
        logger.info("Inserted DataFrame to Cassandra successfully");
    }
}

运行时出现的错误是：

Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: org.apache.spark.sql.cassandra. Please find packages at http://spark.apache.org/third-party-projects.html

你知道吗？

Java cassandra DataFrame apache-spark spark-cassandra-connector

来源：https://stackoverflow.com/questions/60155297/writing-a-dataframe-to-a-cassandra-table-in-java

1条答案

按热度按时间

tpxzln5u1#

您需要确保spark cassandra连接器包含在您提交的结果jar中。
这可以通过构建所谓的fatjar来完成，并提交它。例如，这里是示例（这里是完整的pom）：

...
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <scala.version>2.11.12</scala.version>
    <spark.version>2.4.4</spark.version>
    <spark.scala.version>2.11</spark.scala.version>
    <scc.version>2.4.1</scc.version>
    <java.version>1.8</java.version>
  </properties>
  <dependencies>
     <dependency>
       <groupId>com.datastax.spark</groupId>
       <artifactId>spark-cassandra-connector_${spark.scala.version}</artifactId>
       <version>${scc.version}</version>
     </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_${spark.scala.version}</artifactId>
      <version>${spark.version}</version>
      <scope>provided</scope>
    </dependency>
  </dependencies>
...
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-assembly-plugin</artifactId>
        <version>3.2.0</version>
        <configuration>
          <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
          </descriptorRefs>
        </configuration>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>single</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>

或者您可以指定spark cassandra连接器作为封装通过 --packages com.datastax.spark:spark-cassandra-connector_2.11:2.4.2

展开查看全部

赞(0）回复(0）举报 2021-06-10

我来回答

在java中向cassandra表写入Dataframe

1条答案

相关问题

热门标签

最新问答