MongoDBObject không được thêm vào bên trong vòng lặp rrd foreach casbah scala apache spark

Các tính toán trên RDD được phân phối trên cụm. Bạn không thể cập nhật một biến được tạo bên ngoài quá trình đóng hoạt động RDD từ bên trong RDD. Về cơ bản, chúng ở hai nơi khác nhau:Biến được tạo trong trình điều khiển Spark và được truy cập trong công nhân và nên được coi là chỉ đọc.

Spark hỗ trợ các cummulators phân tán có thể được sử dụng trong trường hợp này: Spark Cummulators

Một tùy chọn khác (tùy chọn tôi thích hơn) là chuyển đổi luồng RDD thành định dạng dữ liệu mong muốn và sử dụng foreachRDD phương pháp duy trì nó vào bộ nhớ thứ cấp. Đây sẽ là một cách hiệu quả hơn để tiếp cận vấn đề. Nó gần giống như thế này:

  val filteredStream = twitterStream.filter(entry => filters.exists(term => entry.getText.getStatus.contains(term)))
  val filteredStreamWithTs = filteredStream.map(x => ((DateTime.now.toString(), x)))
  filteredStreamWithTs.foreachRdd(rdd => // write to Mongo)