Apache Beam에서 JSON 배열을 BigQuery 테이블로 스트리밍하는 방법

여기 내 해결책이 있습니다. JSON 문자열을 List로 한 번 변환 한 다음 c.output을 하나씩 출력합니다. 내 코드는 스칼라에 있지만 Java에서도 같은 일을 할 수 있습니다.

   case class MyTranscationRecord(id: String, name: String, price: Int)
    case class MyTranscation(recordList: List[MyTranscationRecord])
    class ConvertJSONTextToMyRecord extends DoFn[KafkaRecord[java.lang.Long, String], MyTranscation]() {
      private val logger: Logger = LoggerFactory.getLogger(classOf[ConvertJSONTextToMyRecord])
      @ProcessElement
      def processElement(c: ProcessContext): Unit = {
        try {
          val mapper: ObjectMapper = new ObjectMapper()
            .registerModule(DefaultScalaModule)
          val messageText = c.element.getKV.getValue
          val transaction: MyRecord = mapper.readValue(messageText, classOf[MyTranscation])
          logger.info(s"successfully converted to an EPC transaction = $transaction")
          for (record <- transaction.recordList) {
              c.output(record)
          }
        } catch {
          case e: Exception =>
            val message = e.getLocalizedMessage + e.getStackTrace
            logger.error(message)
        }
      }
    }