Откажитесь от плохой записи и загрузите только хорошие записи в фрейм данных из файла json в pyspark

Question

Файл сгенерированный API json выглядит следующим образом. Неправильный формат файла JSON. можем ли мы обрабатывать плохие записи, чтобы отбрасывать и загружать только хорошие строки в фрейм данных, используя pyspark.

{
"name": "PowerAmplifier",
"Component": "12uF Capacitor\n1/21Resistor\n3 Inductor In Henry\PowerAmplifier\n ",
"url": "https://www.onsemi.com/products/amplifiers-comparators/", 
"image": "https://www.onsemi.com/products/amplifiers-comparators/", 
"ThresholdTime": "48min", 
"MFRDate": "2019-05-08", 
"FallTime": "15Min", 
"description": "PowerAmplifier"
}

Код:

spark = SparkSession.builder \
      .master('local') \
      .appName('Fairchild') \
      .config('spark.executor.memory', '2g') \
      .getOrCreate()

df = sqlContext.read.json ("D:\\Fairchild\\component.json").select("name" , "url", "image","ThresholdTime", "MFRDate", "FallTime", "description")
#df.show()
df.registerTempTable("data")
spark.sql("select * from data").show()

Ошибка:

Traceback (most recent call last):
  File "C:/Users/RAJEE/PycharmProjects/onsemi/test.py", line 24, in <module>
    result = spark.sql("select * from data")
  File "C:\spark\python\pyspark\sql\session.py", line 767, in sql
    return DataFrame(self._jsparkSession.sql(sqlQuery), self._wrapped)
  File "C:\spark\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py", line 1257, in __call__
  File "C:\spark\python\pyspark\sql\utils.py", line 69, in deco
    raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: u'java.lang.RuntimeException: java.lang.RuntimeException:

Пожалуйста, поделитесь своим предложением.

Answer 1

Начиная с версии Spark 2.3, запросы из необработанных файлов JSON / CSV запрещены, если указанные столбцы содержат только внутренний столбец поврежденных записей. Я добавил 2 json записей, 1 хорошее и 1 плохое, затем я могу запросить данные. По умолчанию нужна хотя бы одна хорошая запись json. Вы также можете игнорировать все искаженные записи, используя option("mode", "DROPMALFORMED").

Вы также можете использовать option("mode", "DROPMALFORMED") в своем коде python.

sqlContext.read.option("mode", "DROPMALFORMED").json ("D:\\Fairchild\\component.json").select("name" , "url", "image","ThresholdTime", "MFRDate", "FallTime", "description")

JSON data -

{"name": "PowerAmplifier","Component": "12uF Capacitor\n1/21Resistor\n3 Inductor In Henry\PowerAmplifier\n ","url": "https://www.onsemi.com/products/amplifiers-comparators/", "image": "https://www.onsemi.com/products/mplifiers-comparators/", "ThresholdTime": "48min", "MFRDate": "2019-05-08", "FallTime": "15Min", "description": "PowerAmplifier"}
{"name": "PowerAmplifier","Component": "good record","url": "https://www.onsemi.com/products/amplifiers-comparators/", "image": "https://www.onsemi.com/products/mplifiers-comparators/", "ThresholdTime": "48min", "MFRDate": "2019-05-08", "FallTime": "15Min", "description": "PowerAmplifier"}

Код для чтения без игнорирования искаженного , В этом случае вы можете отфильтровать плохие записи, используя столбец "_corrupt_record".

    val j = spark.read.json("/Users/msayed2/Documents/temp/test.txt")
    j.show()

Результат

+-----------+--------+----------+-------------+--------------------+--------------+--------------------+--------------+--------------------+
|  Component|FallTime|   MFRDate|ThresholdTime|     _corrupt_record|   description|               image|          name|                 url|
+-----------+--------+----------+-------------+--------------------+--------------+--------------------+--------------+--------------------+
|       null|    null|      null|         null|{"name": "PowerAm...|          null|                null|          null|                null|
|good record|   15Min|2019-05-08|        48min|                null|PowerAmplifier|https://www.onsem...|PowerAmplifier|https://www.onsem...|
+-----------+--------+----------+-------------+--------------------+--------------+--------------------+--------------+--------------------+

Вы можете полностью игнорировать искаженные данные, как это -

    val j = spark.read.option("mode", "DROPMALFORMED").json("/Users/msayed2/Documents/temp/test.txt")
    j.show()

Результат не искажен

+-----------+--------+----------+-------------+--------------+--------------------+--------------+--------------------+
|  Component|FallTime|   MFRDate|ThresholdTime|   description|               image|          name|                 url|
+-----------+--------+----------+-------------+--------------+--------------------+--------------+--------------------+
|good record|   15Min|2019-05-08|        48min|PowerAmplifier|https://www.onsem...|PowerAmplifier|https://www.onsem...|
+-----------+--------+----------+-------------+--------------+--------------------+--------------+--------------------+