有没有办法将 Spark-cassandra 与 SpringBoot 2 集成？

Question

我正在使用 JHipster 生成应用程序。我正在尝试将火花集成到弹簧靴中。但不知何故，它不起作用。我是弹簧靴和火花的新手。我没有得到任何异常或错误，但也没有得到输出。如果我使用 Java-Spark-Cassandra，它运行良好。谁能告诉我我的代码有什么问题？

//SparkService.java

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

@Service
public class SparkService{

    @Autowired
    private JavaSparkContext javaSparkContext;

    @Autowired
    private SparkSession sparkSession;

    @Value("${spring.data.cassandra.keyspace-name}")
    private String CassandraKeyspace;

    @Value("${cassandra.table}")
    private String CassandraTable;

        public void getAllOrders() {
            try{
            Map<String, String> options = new HashMap<String, String>();
            options.put("keyspace", CassandraKeyspace);
            options.put("table", CassandraTable);

            sparkSession
                .read()
                .format("org.apache.spark.sql.cassandra")
                .options(options)
                .load()
                .createOrReplaceTempView(CassandraTable);
            
            Dataset<Row> df = sparkSession.sql("select * from instruments");
            df.show();

        }
        catch(Exception ex){
        ex.printStackTrace();
    }
    }


}

//SparkServiceImpl.java

package com.celeritio.sparkgateway.spark;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

@Component
public class SparkServiceImpl {
    @Autowired
    SparkService sparkService;

    @EventListener(ContextRefreshedEvent.class)
    public void myMain(){
        System.out.println("myMain");
        sparkService.getAllOrders();
    }
}

//SparkConfiguration.java

@Configuration
@PropertySource("classpath:sparkconfig.properties")

public class SparkConfiguration {

    private static Logger log = LoggerFactory.getLogger(SparkConfiguration.class.getName());

    @Value("${spark.master}")
    private String sparkMaster;

    @Value("${spring.data.cassandra.keyspace-name}")
    private String cassandraKeyspace;

    @Value("${cassandra.table}")
    private String cassandraTable;

    @Value("${spring.data.cassandra.contact-points}")
    private String cassandraHost;

    @Value("${spring.data.cassandra.port}")
    private String cassandraPort;

    @Bean
    public SparkConf sparkConf() {
        SparkConf conf = new SparkConf(true)
            .set("spark.cassandra.connection.host",cassandraHost)
            .set("spark.cassandra.connection.port", cassandraPort)
            .setMaster(sparkMaster)
            .setAppName("SparkConfiguration");
        System.out.println("SparkConf"+conf.isTraceEnabled());
        return conf;
    }

    @Bean
    public JavaSparkContext javaSparkContext() {
        log.info("Connecting to spark with master Url: {}, and cassandra host: {}",
            sparkMaster, cassandraHost);

        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf());

        log.debug("spark context created");

        return javaSparkContext;
    }


    @Bean
    public SparkSession sparkSession() {
        return SparkSession
            .builder()
            .config(sparkConf())
            .sparkContext(javaSparkContext().sc())
            .appName("SparkConfiguration")
            .getOrCreate();

    }

}

Answer 1

我认为问题只是在日志框架中，Spark 使用的是 log4j，而 Spring 使用的是 logback。

我记得在与 Spark 集成时，我们需要从 Spring 中删除 logback 依赖项。