请考虑下列类别：

class MongoDumpService @Inject()(eventsDao: EventDAO)(implicit val ec: ExecutionContext, mat: Materializer) extends LazyLogging {

  private[services] def toAssetsWriterSink: Sink[List[Asset], FileDetails] = ParquetService.toParquetSingleFile[List[Asset]](AppConfig.AssetsFileName)
  private[services] def toExpenseWriterSink: Sink[List[Expense], FileDetails] = ParquetService.toParquetSingleFile[List[Expense]](AppConfig.ExpensesFileName)
  private[services] def toReportsWriterSink: Sink[List[Report], FileDetails] = ParquetService.toParquetSingleFile[List[Report]](AppConfig.ReportsFileName)
  private[services] def toTransactionsWriterSink: Sink[List[Transaction], FileDetails] = ParquetService.toParquetSingleFile[List[Transaction]](AppConfig.TransactionsFileName)
  private[services] def toEventsWriterSink: Sink[PacificOriginalEvent, FileDetails] = ParquetService.toParquetSingleFile[PacificOriginalEvent](AppConfig.PacificOriginalEventFileName)

  def createMongoDump(recordingId: BSONObjectID, maxDocs: Option[Int] = None): List[FileDetails] = RunnableGraph.fromGraph(
    GraphDSL.create(toAssetsWriterSink, toExpenseWriterSink, toReportsWriterSink, toTransactionsWriterSink, toEventsWriterSink, sharedKillSwitch.flow[Event])((f1,f2,f3,f4,f5,_) => List(f1,f2,f3,f4,f5)) {
      import GraphDSL.Implicits._
      implicit builder =>
        (writeAssets, writeExpenses, writeReports, writeTransactions, writerEvents, sw) =>

          val source    = builder.add(eventsDao.getEventsSource(recordingId.stringify, maxDocs))
          val broadcast = builder.add(Broadcast[Event](5))

          source ~> sw ~> broadcast
                          broadcast.out(Write.PacificEvents).map(_.pacificEvent)                                     ~> writerEvents
                          broadcast.out(Write.Expenses).filter(_.expenses.isDefined).map(_.expenses.get)             ~> writeExpenses
                          broadcast.out(Write.Assets).filter(_.assets.isDefined).map(_.assets.get)                   ~> writeAssets
                          broadcast.out(Write.Reports).filter(_.reports.isDefined).map(_.reports.get)                ~> writeReports
                          broadcast.out(Write.Transactions).filter(_.transactions.isDefined).map(_.transactions.get) ~> writeTransactions

          ClosedShape

    }).run()

}

此代码是返回List[FileDetails]]，其实际将包含Option[List[T]]的一些字段的Event Object写入其应写入的文件，例如fieldA ~> writerFieldA等问题如下：
我想等待直到此操作完成，因为这将上载到S3文件，文件名为0KB：

private[actors] def uploadDataToS3(recording: Recording) = {
    logger.info(s"Uploading data to S3 with recordingId: ${recording._id.stringify}")
    val details = mongoDumpService.createMongoDump(recording._id, recording.limit)
    s3Service.uploadFiles(recording._id.stringify, details)
  }

如果没有图形DSL，我可以执行runWith，返回Future[..]

如何使用graphDSL实现这一点？（我希望返回Future[List[FileDetails]]]

编辑：

已添加到ParquetSingleFile

def toParquetSingleFile[In](fileName: String)(implicit
                                                ec: ExecutionContext,
                                                mat: Materializer,
                                                writes: Writes[In]): Sink[In, FileDetails] = {
    val absolutePath = TEMP_DIRECTORY + File.separator + s"$fileName.${FileExtension.PARQUET.toSuffix}"
    toJsString[In]
      .log(s"ParquetService", _ => s"[✍️] - Writing element toParquetSingleFile for path: $absolutePath ...")
      .withAttributes(Attributes.logLevels(onFailure = LogLevels.Error, onFinish = LogLevels.Off, onElement = LogLevels.Info))
      .to(
        ParquetStreams.toParquetSingleFile(
          path = absolutePath,
          options = ParquetWriter.Options(
            writeMode = ParquetFileWriter.Mode.OVERWRITE,
            compressionCodecName = CompressionCodecName.GZIP))
      ).mapMaterializedValue(_ => FileDetails(absolutePath, FileExtension.PARQUET))
  }

解决方案：

def toParquetSingleFile[In](fileName: String)(implicit ec: ExecutionContext, mat: Materializer, writes: Writes[In]): Sink[In, Future[Option[FileDetails]]] = {
    val absolutePath = TEMP_DIRECTORY + File.separator + s"$fileName.${FileExtension.PARQUET.toSuffix}"
    toJsString[In]
      .toMat(
        Sink.lazySink(() => ParquetStreams.toParquetSingleFile(
          path = absolutePath,
          options = ParquetWriter.Options(
            writeMode = ParquetFileWriter.Mode.OVERWRITE,
            compressionCodecName = CompressionCodecName.GZIP))
        )
      )(Keep.right)
      .mapMaterializedValue(_.flatten
        .map { _ =>
          logger.info(s"[ParquetService] - [✍️] Writing file: [$absolutePath] Finished!")
          Some(FileDetails(absolutePath, FileExtension.PARQUET))
        }
        .recover {
          case _: NeverMaterializedException => Option.empty[FileDetails]
        }
      )
  }