【发布时间】：2019-11-20 08:40:02
【问题描述】：
我正在从事一个数据分析项目。我是 Spark 的新手。我创建了一个 SpringBoot 项目，在该项目中我使用了一些使用来自 Kafka 主题的数据的 Spark Streaming 消费者。
我在同一个 SpringBoot 应用程序中使用了所有组件，例如 Kafka、Spark。下面是 SpringBoot Main 类，我在其中使用命令行运行器初始化 Spark Streaming 作业。
@SpringBootApplication
@EnableCaching
public class SpringApplication
{
public static void main(String[] args)
{
SpringApplication.run(SpringApplication.class, args);
}
@Bean
public EnrichEventSparkConsumerRunner sparkEnrichEventConsumerRunner()
{
return new EnrichEventSparkConsumerRunner();
}
@Bean
public RawEventSparkConsumerRunner sparkRawEventConsumerRunner()
{
return new RawEventSparkConsumerRunner();
}
}
public class EnrichEventSparkConsumerRunner implements CommandLineRunner
{
@Autowired
JavaStreamingContext javaStreamingContext;
@Autowired
EnrichEventSparkConsumer enrichEventSparkConsumer;
@Override
public void run(String... args) throws Exception
{
// start Raw Event Spark Cosnumer.
JobContextImpl jobContext = new JobContextImpl(javaStreamingContext);
// start Enrich Event Spark Consumer.
enrichEventSparkConsumer.startEnrichEventConsumer(jobContext.streamingctx());
}
}
public class RawEventSparkConsumerRunner implements CommandLineRunner
{
@Autowired
JavaStreamingContext javaStreamingContext;
@Autowired
RawEventSparkConsumer rawEventSparkConsumer;
@Override
public void run(String... args) throws Exception
{
// start Raw Event Spark Cosnumer.
JobContextImpl jobContext = new JobContextImpl(javaStreamingContext);
rawEventSparkConsumer.sparkRawEventConsumer(jobContext.streamingctx());
}
}
RawEventSparkConsumer.java - 在这个 SparkStreaming 类中，我们正在使用来自 Kafka 主题的数据，丰富数据并将该数据保存到弹性中，并且我们将这些丰富的数据发送到下一个 Kafka 主题，即由 EnrichEventSparkConsumer 消费。
@Component
public class RawEventSparkConsumer implements Serializable
{
private final Logger logger = LoggerFactory.getLogger(RawEventSparkConsumer.class);
@Autowired
private ElasticSearchServiceImpl dataModelServiceImpl;
@Autowired
private EnrichEventKafkaProducer enrichEventKafkaProducer;
@Autowired
private SparkConfiguration sparkConfiguration;
@Autowired
private RawAttributesConfig rawAttConfig;
private static ObjectMapper mapper;
public void sparkRawEventConsumer(JavaStreamingContext streamingContext)
{
mapper = new ObjectMapper();
Collection<String> topics = Arrays.asList(sparkConfiguration.getRawEventTopic());
Map<String, Object> kafkaParams = new HashedMap();
kafkaParams.put("bootstrap.servers", sparkConfiguration.getBootStrapServers());
kafkaParams.put("key.deserializer", StringDeserializer.class);
kafkaParams.put("value.deserializer", StringDeserializer.class);
kafkaParams.put("group.id", "group1");
kafkaParams.put("auto.offset.reset", "latest");
kafkaParams.put("enable.auto.commit", true);
JavaInputDStream<ConsumerRecord<String, String>> rawEventRDD = KafkaUtils.createDirectStream(streamingContext,
LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
JavaDStream<String> dStream = rawEventRDD.map((x) -> x.value());
/*
* JavaDStream<EnrichEventDataModel> enrichEventRdd = dStream.map((raw) -> { BaseDataModel csvDataModel =
* mapper.readValue(raw, BaseDataModel.class); return new EnrichEventDataModel(csvDataModel); });
*/
JavaDStream<List<Map<String, Object>>> enrichEventRdd = dStream.map(convertIntoMapList);
enrichEventRdd.foreachRDD(rdd -> {
logger.info("Inside rawEventRDD.foreachRDD = = = " + rdd.count());
sendEnrichEventToKafkaTopic(rdd.collect());
});
streamingContext.start();
try {
streamingContext.awaitTermination();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
logger.info("RawEvent consumer SparkStreaming job started.");
}
HashMap<String, UserIndexDto> userMap = new HashMap();
private void sendEnrichEventToKafkaTopic(List<List<Map<String, Object>>> list)
{
if (enrichEventKafkaProducer != null && list != null && list.size() > 0)
try {
logger.info("sendEnrichEventToKafkaTopic, csv raw log count: "+ list.size());
list.parallelStream().forEach(mapList -> {
logger.info("sendEnrichEventToKafkaTopic -- mapListmapList, mapList: "+ mapList.size());
if (Objects.nonNull(mapList) && mapList.size() > 0) {
List<Map<String, Object>> enrichedMapList = mapList.parallelStream().map(mapData -> {
if (mapData.containsKey(rawAttConfig.getAccountname())) {
String accountName = String.valueOf(mapData.get(rawAttConfig.getAccountname()));
if(accountName != null) {
accountName = accountName.trim();
}
if (accountName != null && accountName.length() > 0 && !userMap.containsKey(accountName)) {
accountName = accountName.split("@")[0];
List<User> userList = dataModelServiceImpl.getUser(accountName,"u_employeeId");
if (userList != null && userList.size() > 0) {
User user = userList.get(0);
if(!user.getU_userId().equalsIgnoreCase(accountName)) {
List<User> userList1 = dataModelServiceImpl.getUser(accountName,"u_email");
if(userList1 != null && userList1.size() > 0) {
User user1 = userList1.get(0);
UserIndexDto userdto1 = new UserIndexDto();
userdto1.setUserId(user1.getU_email());
userdto1.setEmpId(user1.getU_employeeId());
userMap.put(userdto1.getUserId(), userdto1);
}
}else {
UserIndexDto userdto = new UserIndexDto();
userdto.setUserId(user.getU_userId());
userdto.setEmpId(user.getU_employeeId());
userMap.put(userdto.getUserId(), userdto);
}
writeToLogsFile("Enriching RawEvent using ElasticIndex for accountName="
+ userList.get(0).getU_employeeId(), Constants.INFO);
}else {
List<User> userList1 = dataModelServiceImpl.getUser(accountName,"u_email");
if(userList1 != null && userList1.size() > 0) {
User user1 = userList1.get(0);
UserIndexDto userdto1 = new UserIndexDto();
userdto1.setUserId(user1.getU_email());
userdto1.setEmpId(user1.getU_employeeId());
userMap.put(userdto1.getUserId(), userdto1);
}
}
}
UserIndexDto userdto = userMap.get(accountName);
mapData.put("userId", userdto != null ? userdto.getUserId() : null );
mapData.put("empId", userdto != null ? userdto.getEmpId() : null);
// writeToLogsFile("Enriching RawEvent using Map for accountName=" + (userdto != null ? userdto.getUserId() : ""),
// Constants.INFO);
} else {
mapData.put("userId", null);
mapData.put("empId", null);
}
mapData.put("enrichEventId", UUID.randomUUID().toString());
saveDataToEasticSearch(mapData);
return mapData;
}).collect(Collectors.toList());
//saveDataToElasticSearch(enrichedMapList);
processEnrichEvents(enrichedMapList);
}
});
} catch (Exception e) {
writeToLogsFile(e.getMessage(), Constants.ERROR);
}
}
private void saveDataToEasticSearch(Map<String, Object> mapData) {
if(Objects.nonNull(mapData)) {
dataModelServiceImpl.saveEnrichModel(mapData);
}
}
static Function convertIntoMapList = new Function<String, List<Map<String, Object>>>()
{
@Override
public List<Map<String, Object>> call(String raw) throws Exception
{
// TODO Auto-generated method stub
return mapper.readValue(raw, new TypeReference<List<HashMap<String, Object>>>(){});
}
};
private void processEnrichEvents(List<Map<String, Object>> enrichedMapList)
{
try {
Thread.sleep(3000);
enrichEventKafkaProducer.sendEnrichEvent(enrichedMapList);
} catch (Exception e) {
e.printStackTrace();
}
}
private void saveDataToElasticSearch(List<Map<String, Object>> enrichedMapList)
{
if (!enrichedMapList.isEmpty())
dataModelServiceImpl.saveAllEnrichModel(enrichedMapList);
}
private void writeToLogsFile(String message, String loglevel)
{
if (loglevel.equalsIgnoreCase(Constants.ERROR)) {
logger.error(message);
} else if (loglevel.equalsIgnoreCase(Constants.INFO)) {
logger.info(message);
}
}
现在我使用 mvn clean package 命令为其创建了一个 jar，并使用 java -jar 命令运行该 jar，该命令使用命令行运行程序初始化 Spark 作业。即使我们使用它进行 POC，它也能正常工作。
现在我们要在集群环境中运行 Spark。所以我有一些与之相关的疑问。我进行了很多搜索，但没有找到正确的方法。以下是问题和疑问 -
- 我们通过命令行运行程序使用 Spark Streaming 是否正确？
- 当我阅读运行集群环境时，我们必须使用 spark-submit 命令提交 spark streamign。如何 我可以在我当前的项目中做吗？问题是，据我所知，对于 Spark-submit，我们需要一个主要方法。所以 我需要对 Spark-Submit 进行哪些更改。
- 我们正在使用弹性 API 将数据保存到弹性中。这是正确的做法吗？
- 请查看这一行 sendEnrichEventToKafkaTopic(rdd.collect()); 我们在 rdd 上调用 .collect。我读到，当我们调用 .collect 时，这个东西在主节点上运行。所以我们必须避免.collect。那么其他的方法是什么。
【问题讨论】：
标签： spring-boot apache-spark apache-kafka spark-streaming