将 FASTQ 文件读入 AWS Glue 作业脚本

Posted 2023-04-15

【中文标题】将 FASTQ 文件读入 AWS Glue 作业脚本

【英文标题】：Read FASTQ file into a AWS Glue Job Script

【发布时间】：2020-09-30 14:47:26

【问题描述】：

我需要，但出现此错误：

Traceback（最近一次调用最后）：文件“/opt/amazon/bin/runscript.py”，第 59 行，在 runpy.run_path(script, run_name='ma​​in') 文件“ /usr/lib64/python3.7/runpy.py”，第 261 行，在 run_path 代码，fname = _get_code_from_file(run_name, path_name) 文件“/usr/lib64/python3.7/runpy.py”，第 236 行，在 _get_code_from_file code = compile(f.read(), fname, 'exec') File "/tmp/test20200930", line 24 datasource0 = spark.createDataset(sc.textFile("s3://sample-genes-data/fastq/S_Sonnei_short_reads_1 .fastq").sliding(4, 4).map ^ SyntaxError: invalid syntax 在处理上述异常的过程中，发生了另一个异常： Traceback (most recent call last): File "/opt/amazon/bin/runscript.py", line 92, in 而new_stack.tb_frame.f_code.co_filename中的“runpy.py”：AttributeError：'NoneType'对象没有属性'tb_frame'

这是我的代码：

import org.apache.spark.mllib.rdd.RDDFunctions._

datasource0 = spark.createDataset(sc.textFile("s3://sample-genes-data/fastq/S_Sonnei_short_reads_1.fastq").sliding(4, 4).map 
  case Array(id, seq, _, qual) => (id, seq, qual)
 ).toDF("identifier", "sequence", "quality")
datasource1 = DynamicFrame.fromDF(datasource0, glueContext, "nullv")

我点击了这个链接： Read FASTQ file into a Spark dataframe

【参考方案1】：

我能够通过将代码包装在 GlueApp 对象中来运行代码。您可以通过替换您的 S3 路径来使用以下代码。

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import org.apache.spark.SparkContext
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.SparkSession
import com.amazonaws.services.glue.DynamicFrame
import org.apache.spark.mllib.rdd.RDDFunctions._

object GlueApp 
  def main(sysArgs: Array[String]) 
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    val sparkSession: SparkSession = glueContext.getSparkSession
    import sparkSession.implicits._
    val datasource0 = sparkSession.createDataset(spark.textFile("s3://<s3path>").sliding(4, 4).map 
  case Array(id, seq, _, qual) => (id, seq, qual)
 ).toDF("identifier", "sequence", "quality")
   val datasource1 = DynamicFrame(datasource0, glueContext)
   datasource1.show()
   datasource1.printSchema()
   Job.commit()
  

通过的输入：

@seq1
AGTCAGTCGAC
+
?@@FFBFFDDH
@seq2
CCAGCGTCTCG
+
?88ADA?BDF8

输出：

"identifier": "@seq1", "sequence": "AGTCAGTCGAC", "quality": "?@@FFBFFDDH"
"identifier": "@seq2", "sequence": "CCAGCGTCTCG", "quality": "?88ADA?BDF8"

【讨论】：

我收到这个错误：Traceback (last last call last): object GlueApp ^ SyntaxError: invalid syntax 在处理上述异常期间

您在胶水作业中使用的是 pyspark 还是 scala 代码？

您发布的示例是 scala one，您需要在创建 Glue 作业时选择 scala 作为编程语言