Блог

Я новичок в Spark-Java , я хочу получить подсписок из списка в Java 8 . Затем я конвертирую его в RDD. Я сделал это в приведенном ниже коде:

             List<CSVRecord> inputRecords = readInputLayer(actorSystem, inputCatalog, inputCatalogVersion);
            LOGGER.info("Number of partition "  inputRecords.size());

            List<CSVRecord> inputRecordsTmp = inputRecords.stream().limit(100).collect(Collectors.toList());


            JavaRDD<CSVRecord> inputRecordsJavaRDD = JavaSparkContext.emptyRDD();
            for (List<CSVRecord> partition: inputRecordsTmp ){
                JavaRDD<CSVRecord> inputRecordsTmpRDD = (JavaRDD<CSVRecord>) JavaSparkContext.parallelize(partition);
                inputRecordsJavaRDD = JavaSparkContext.union(inputRecordsJavaRDD,inputRecordsTmpRDD);

            }

        LOGGER.info("Number of lines to insert JAVA RDD ="  inputRecordsJavaRDD.count());
  

Но я получил ошибку в цикле для,
он не принимает List<CSVRecord> partition:

 Incompatible types:
Required: org.apache.commons.csvRecord
Found: java.util.list  <org.apache.commons.csvRecord>
  

Как я могу это исправить?
Спасибо

РЕДАКТИРОВАТЬ: Просто взглянул на документацию. parallelize фактически принимает список, но это означает только то, что вам вообще не нужен foreach. Вы должны быть в состоянии просто передать свои inputRecordsTmp права в parallelize и извлечь из этого свой RDD.

Тем не менее, поскольку foreach сформулирован неправильно, я оставлю это здесь для получения дополнительной информации:

Прежде всего, ваш цикл for-each не выглядит для меня абсолютно правильным. При написании цикла for-each-loop на Java предполагается, что часть перед двоеточием относится к типу элемента в коллекции, которую вы перебираете.

В вашем случае у вас есть список, содержащий объекты CSVRecord. В for-each вы в основном говорите «для каждого CSVRecord в этом списке объектов CSVRecord …». Однако вы написали «для каждого списка объектов CSVRecord в этом списке объектов CSVRecord …», что не имеет особого смысла.

Java уже знает, что inputRecordsTmp это список CSVRecords, поэтому нет необходимости указывать это снова где-либо. Вместо этого вы хотите сообщить, что для каждого объекта в списке вы хотите извлечь этот объект в переменную ( partition в вашем случае), чтобы вы могли использовать этот извлеченный объект внутри вашего цикла.

Поэтому вместо того, чтобы писать

 for (List<CSVRecord> partition: inputRecordsTmp ){
  

Вам пришлось бы написать

 for (CSVRecord record: inputRecordsTmp ){
  

Это может решить вашу проблему. Если нет, не могли бы вы, пожалуйста, добавить информацию о том, в какой именно строке происходит сбой?