У меня возникла проблема при попытке прочитать более 40 ГБ данных CSV из S3 Athena.
2024-06-25 18:16:25.447 ПРЕДУПРЕЖДЕНИЕ 509917 --- [pool-6-thread-1] c.a.s.s.internal.S3AbortableInputStream : не все байты были прочитаны из S3ObjectInputStream, что прерывает HTTP-соединение. Вероятно, это ошибка, которая может привести к неоптимальному поведению. Запрашивайте только те байты, которые вам нужны, с помощью GET с диапазоном или сливайте входной поток после использования.
private static int saveResultFile(S3Object s3Object, String outputFile, List columnInfoList) {
System.out.println("This is the save file method.");
System.out.println(outputFile + "this is output");
String lastTwoDigits = outputFile.substring(outputFile.length() - 2);
boolean hasData = false; // Initialize to false
int totalSum = 0; // Initialize totalSum before the loop
// Set the maximum records per page
int maxRecordsPerPage = 1000000;
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(s3Object.getObjectContent(), StandardCharsets.UTF_8));
ZipOutputStream zipOutputStream = new ZipOutputStream(new FileOutputStream(outputFile + ".zip"))) {
CSVReader csvReader = new CSVReader(reader);
String[] header = csvReader.readNext();
if (header == null) {
// No header, return false
System.out.println("No header found in the CSV file.");
return NO_HEADERS_FOUND;
}
int recordsProcessed = 0;
int pageNumber = 1;
CSVWriter writer = null;
try {
// Create a ZipOutputStream to write to the zip file
writer = new CSVWriter(new FileWriter(outputFile + "_" + pageNumber + ".csv"), ',',
CSVWriter.DEFAULT_QUOTE_CHARACTER);
// Write the header for the current page
writer.writeNext(header);
String[] line;
while ((line = csvReader.readNext()) != null) {
hasData = true; // Set to true if any data is processed
if (lastTwoDigits.equals("02")) {
// Calculate sum for sms_split column
if (line.length > 8) { // Assuming index 7 is the 8th column
try {
totalSum += Integer.parseInt(line[8]); // Assuming sms_split is an integer
} catch (NumberFormatException e) {
// Handle parsing error if necessary
e.printStackTrace();
}
}
}
int lastColumnIndex = line.length - 1;
String lastColumnName = header[lastColumnIndex];
if (isTextColumn(lastColumnName)) {
// Perform encryption and decryption operations only if it's a text column
String encryptedValue = line[lastColumnIndex];
String decryptedText = decrypt(encryptedValue);
if (isHexString(decryptedText)) {
decryptedText = hexToAscii(decryptedText);
}
line[lastColumnIndex] = decryptedText;
}
if (recordsProcessed >= maxRecordsPerPage) {
// Close the current writer and create a new one for the next page
writer.close();
pageNumber++;
recordsProcessed = 0;
writer = new CSVWriter(new FileWriter(outputFile + "_" + pageNumber + ".csv"), ',',
CSVWriter.DEFAULT_QUOTE_CHARACTER);
// Write the header for the new page
writer.writeNext(header);
}
writer.writeNext(line);
recordsProcessed++;
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("Error creating ZIP file: " + outputFile + ".zip");
} finally {
if (writer != null) {
writer.close();
}
}
if (hasData) {
for (int i = 1; i = 0) {
zipOutputStream.write(bytes, 0, length);
}
zipOutputStream.closeEntry();
if (new File(pageFileName).delete()) {
// System.out.println("Deleted file: " + pageFileName);
} else {
System.out.println("Failed to delete file: " + pageFileName);
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("Error processing file: " + pageFileName);
}
}
System.out.println("Result files saved successfully with decryption.");
System.out.println("Total sum of sms_split column: " + totalSum);
} else {
System.out.println("No data found to save.");
return NO_HEADERS_FOUND;
}
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (s3Object != null) {
s3Object.close();
}``your text``
} catch (IOException e) {
e.printStackTrace();
}
}
return totalSum;
}
Подробнее здесь: https://stackoverflow.com/questions/787 ... pring-boot
