E-MapReduce:よくある質問

Spark の履歴ジョブはどこで確認できますか?

EMR コンソールの対象クラスタの [アクセスリンクとポート] タブにある Spark UI リンクをクリックすると、Spark の履歴ジョブの実行情報を確認できます。UI へのアクセスの詳細については、「コンソールからオープンソースコンポーネントの Web インターフェースにアクセスする」をご参照ください。

スタンドアロンモードは Spark ジョブの送信でサポートされていますか?

いいえ。E-MapReduce は、YARN 上の Spark と Kubernetes 上の Spark モードを使用したジョブの送信をサポートしていますが、スタンドアロンモードまたは Mesos モードはサポートしていません。

Spark2 コマンドラインツールのログ出力を削減するにはどうすればよいですか?

Spark3 のスモールファイルマージ機能を使用するにはどうすればよいですか?

パラメーター spark.sql.adaptive.merge.output.small.files.enabled を true に設定することで、小さなファイルを自動的にマージできます。マージされたファイルは圧縮されるため、マージされたファイルが小さすぎる場合は、パラメーター spark.sql.adaptive.advisoryOutputFileSizeInBytes の値を適切に増やすことができます。デフォルト値は 256 MB です。

SparkSQL でデータスキューを処理するにはどうすればよいですか?

• Spark2 の場合、処理方法は次のとおりです。

  • テーブルを読み取るときに、null などの無関係なデータをフィルタリングします。

  • 小さなテーブルをブロードキャストします。

    select /*+ BROADCAST (table1) */ * from table1 join table2 on table1.id = table2.id

  • スキューキーに基づいてスキューデータを分離します。

    select * from table1_1 join table2 on table1_1.id = table2.id
    union all
    select /*+ BROADCAST (table1_2) */ * from table1_2 join table2 on table1_2.id = table2.id

  • スキューキーがわかっている場合は、データを分散させます。

    select id, value, concat(id, (rand() * 10000) % 3) as new_id from A
    select id, value, concat(id, suffix) as new_id
    from ( 
    select id, value, suffix
     from B Lateral View explode(array(0, 1, 2)) tmp as suffix)

  • スキューキーがわからない場合は、データを分散させます。

    select t1.id, t1.id_rand, t2.name
    from (
    select id ,
    case when id = null then concat(‘SkewData_’, cast(rand() as string))
    else id end as id_rand
    from test1
    where statis_date = ‘20221130’) t1
    left join test2 t2
    on t1.id_rand = t2.id

• Spark3 の場合、EMR コンソールの Spark3 サービスの [構成] タブで、Spark.sql.adaptive.enabled と spark.sql.adaptive.skewJoin.enabled のパラメーター値を true に変更できます。

PySpark に Python 3 バージョンを指定するにはどうすればよいですか?

以下の内容は、Spark2 をオプションサービスとして、DataLake クラスタの EMR-5.7.0 バージョンを例として、PySpark に Python 3 バージョンを指定する方法を紹介します。

次の 2 つの方法で Python バージョンを変更できます。

Spark Streaming ジョブがしばらく実行された後に予期せず終了するのはなぜですか?

• まず、Spark のバージョンが 1.6 より前かどうかを確認します。そうである場合は、Spark のバージョンを更新してください。

  Spark 1.6 より前のバージョンにはメモリリークの問題があり、コンテナが中止される可能性があります。

• コードがメモリ使用量に関して最適化されているかどうかを確認します。

Spark Streaming ジョブが終了したにもかかわらず、E-MapReduce コンソールにジョブステータスが「実行中」と表示されるのはなぜですか?

ジョブの送信方法が Yarn クライアントモードかどうかを確認します。E-MapReduce には Yarn クライアントモードでの Spark Streaming ジョブのステータスの監視に問題があるため、Yarn クラスタモードに変更してください。

Kerberos 認証が有効になっているクラスタで YARN クラスタモードで spark-submit を実行すると、エラー java.lang.ClassNotFoundException が発生するのはなぜですか?

具体的なエラーメッセージは以下に示されています。

原因分析: EMR クラスタで Kerberos が有効になっていると、YARN クラスタモードのドライバのクラスパスは指定されたディレクトリにある JAR ファイルを自動的に含むように拡張されないため、Spark タスクの実行時にエラーが発生します。

解決策: EMR クラスタで Kerberos が有効になっている場合、YARN クラスタモードで spark-submit を使用してタスクを送信するときは、--jars パラメーターを追加する必要があります。ユーザープログラム自体が依存している JAR パッケージに加えて、/opt/apps/METASTORE/metastore-current/hive2 ディレクトリにあるすべての JAR パッケージを追加する必要もあります。

たとえば、アプリケーション JAR パッケージが /opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar の場合、spark-submit コマンドは次のとおりです。

spark-submit --deploy-mode cluster --class org.apache.spark.examples.SparkPi --master yarn \
--jars $(ls /opt/apps/METASTORE/metastore-current/hive2/*.jar | tr '\n' ',') \
/opt/apps/SPARK3/spark3-current/examples/jars/spark-examples_2.12-3.5.3-emr.jar