Общий обзор моей цели: мне нужно найти файл (ы) (они в формате JSON), которые содержат определенный идентификатор.В основном нужно вернуть DF (или список) идентификатора и имя файла, который его содержит.
// Read in the data from s3
val dfLogs = spark.read.json("s3://some/path/to/data")
.withColumn("fileSourceName", input_file_name())
// Filter for the ID and select then id and fileSourceName
val results = dfLogs.filter($"id" === "some-unique-id")
.select($"id", $"fileSourceName")
// Return the results
results.show(false)
Звучит достаточно просто, верно?Однако проблема, с которой я сталкиваюсь, состоит в том, что каталог S3, из которого я читаю, содержит миллионы (приблизительно 5+ миллионов) файлов, средний размер которых составляет 10 КБ.Маленькая проблема с файлами!Чтобы сделать это, я раскручивал кластер из 5 узлов (m4.xlarge) в EMR и использовал Zeppelin для интерактивного запуска приведенного выше кода.
Тем не менее, я продолжаю получать следующее сообщение об ошибке при выполнении первого оператора spark (чтение):
org.apache.thrift.transport.TTransportException at
org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:132)
Мне трудно узнать больше об этой ошибке, но яподозреваю, что это связано с запросами от моей искровой работы к s3.
Есть ли у кого-нибудь предложения о том, как обрабатывать так много маленьких файлов?Должен ли я выполнить s3-dist-cp из S3 -> HDFS в кластере EMR, а затем выполнить приведенный выше запрос, но прочитать из HDFS?Или какой-то другой вариант?Это одноразовое занятие ... стоит ли создавать супер большой кластер?Это улучшит производительность или решит мою ошибку?Я думал о попытке сгруппировать файлы в более крупные ... но мне нужны уникальные файлы, в которых содержится идентификатор.
Я хотел бы изменить способ агрегирования этих файлов вS3 ... но я ничего не могу с этим поделать.
Примечание: я видел несколько сообщений здесь, но они довольно старые .Другая ссылка, но это, я не думаю, относится к моей ситуации