Handling bad records and files | Databricks on AWS [2022/5/16時点]の翻訳です。
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
ファイルベースのデータソースからデータを読み込む際、Apache Spark SQLは2つの典型的なエラーに遭遇します。第一にファイルを読み込めない場合(例えば、ファイルが存在しない、アクセスできない、破損している)があります。第二に、ファイルが処理可能であったとしても、いくつかのレコードがパースできない場合(例えば、構文エラーやスキーマのミスマッチ)があります。
Databricksでは、Sparkジョブを阻害することなしに不正なレコードやファイルを取り扱うための統合インタフェースを提供しています。データソースのオプションbadRecordsPathを指定することで、例外のログから例外レコード/ファイルや理由を取得することができます。badRecordsPathでは、CSVやJSONソースにおける不正レコードに関する情報やすべてのファイルベースのビルトインソース(Parquetなど)の不正ファイルを記録する例外ファイルのパスを指定します。
さらに、ファイルを読み込む際、ネットワーク接続例外、IO例外などの一時的なエラーが起きる場合があります。これらのエラーは無視されますが、badRecordsPathで記録され、Sparkはタスクの実行を継続します。
注意
ファイルベースのデータソースにおけるbadRecordsPathオプションの使用においては、いくつか重要な制限が存在します。
- 非トランザクションであり、一貫性の無い結果につながる場合があります。
- 一時的なエラーは失敗として取り扱われます。
サンプル
入力ファイルが見つからない
val df = spark.read
.option("badRecordsPath", "/tmp/badRecordsPath")
.format("parquet").load("/input/parquetFile")
// Delete the input parquet file '/input/parquetFile'
dbutils.fs.rm("/input/parquetFile")
df.show()
上のサンプルでは、df.show()が入力ファイルを見つけ出せないため、Sparkはエラーを記録するためにJSONフォーマットで例外ファイルを作成します。例えば、/tmp/badRecordsPath/20170724T101153/bad_files/xyzが例外ファイルのパスとします。このファイルは指定されたbadRecordsPathディレクトリに格納されます。20170724T101153はこのDataFrameReaderの作成時刻です。bad_filesは例外のタイプであり、xyzが不正ファイルのパスと例外の理由のメッセージを含むJSONレコードを格納するファイルとなります。
入力ファイルに不正レコードが含まれている
// Creates a json file containing both parsable and corrupted records
Seq("""{"a": 1, "b": 2}""", """{bad-record""").toDF().write.format("text").save("/tmp/input/jsonFile")
val df = spark.read
.option("badRecordsPath", "/tmp/badRecordsPath")
.schema("a int, b int")
.format("json")
.load("/tmp/input/jsonFile")
df.show()
この例では、データフレームにはパース可能な一行目({"a": 1, "b": 2})が含まれています。2行目の不正なレコード({bad-record)は/tmp/badRecordsPath/20170724T114715/bad_records/xyzのJSON例外ファイルとして記録されます。この例外ファイルには、不正レコード、レコードを含むファイルへのパス、例外の理由に関するメッセージが含まれます。例外ファイルを特定したら、処理するためにJSONリーダーをしようすることができまs