SparkSQL-クイックガイド

業界は、データセットを分析するためにHadoopを幅広く使用しています。その理由は、Hadoopフレームワークが単純なプログラミングモデル（MapReduce）に基づいており、スケーラブルで柔軟性があり、フォールトトレラントで費用効果の高いコンピューティングソリューションを可能にするためです。ここでの主な関心事は、クエリ間の待機時間とプログラムの実行待機時間の観点から、大規模なデータセットの処理速度を維持することです。

Sparkは、Hadoop計算コンピューティングソフトウェアプロセスを高速化するためにApache SoftwareFoundationによって導入されました。

一般的な信念に反して、 Spark is not a modified version of Hadoop独自のクラスター管理があるため、実際にはHadoopに依存していません。Hadoopは、Sparkを実装する方法の1つにすぎません。

SparkはHadoopを2つの方法で使用します–1つは storage そして2番目は processing。Sparkには独自のクラスター管理計算があるため、Hadoopはストレージ目的でのみ使用されます。

Apache Spark

Apache Sparkは、高速計算用に設計された超高速クラスターコンピューティングテクノロジーです。これはHadoopMapReduceに基づいており、MapReduceモデルを拡張して、インタラクティブなクエリやストリーム処理など、より多くの種類の計算に効率的に使用できるようにします。Sparkの主な機能はin-memory cluster computing これにより、アプリケーションの処理速度が向上します。

Sparkは、バッチアプリケーション、反復アルゴリズム、インタラクティブクエリ、ストリーミングなど、幅広いワークロードをカバーするように設計されています。それぞれのシステムでこれらすべてのワークロードをサポートするだけでなく、個別のツールを維持する管理上の負担を軽減します。

ApacheSparkの進化

Sparkは、2009年にカリフォルニア大学バークレー校のAMPLabでMateiZahariaによって開発されたHadoopのサブプロジェクトの1つです。2010年にBSDライセンスの下でオープンソース化されました。2013年にApacheソフトウェア財団に寄付され、現在、ApacheSparkは2014年2月からトップレベルのApacheプロジェクトになりました。

ApacheSparkの機能

ApacheSparkには次の機能があります。

  • Speed− Sparkは、Hadoopクラスターでアプリケーションを実行するのに役立ちます。メモリでは最大100倍、ディスクで実行すると10倍高速になります。これは、ディスクへの読み取り/書き込み操作の数を減らすことで可能になります。中間処理データをメモリに保存します。

  • Supports multiple languages− Sparkは、Java、Scala、またはPythonの組み込みAPIを提供します。したがって、さまざまな言語でアプリケーションを作成できます。Sparkには、インタラクティブクエリ用の80個の高レベル演算子が用意されています。

  • Advanced Analytics− Sparkは、「Map」と「reduce」をサポートするだけではありません。また、SQLクエリ、ストリーミングデータ、機械学習（ML）、グラフアルゴリズムもサポートしています。

SparkはHadoop上に構築されています

次の図は、Hadoopコンポーネントを使用してSparkを構築する3つの方法を示しています。

以下で説明するように、Sparkのデプロイには3つの方法があります。

  • Standalone− Sparkスタンドアロン展開とは、SparkがHDFS（Hadoop分散ファイルシステム）の最上位を占め、HDFSに明示的にスペースが割り当てられることを意味します。ここでは、SparkとMapReduceが並行して実行され、クラスター上のすべてのSparkジョブをカバーします。

  • Hadoop Yarn− Hadoop Yarnのデプロイとは、単純に、事前インストールやルートアクセスを必要とせずにYarnでSparkが実行されることを意味します。SparkをHadoopエコシステムまたはHadoopスタックに統合するのに役立ちます。これにより、他のコンポーネントをスタック上で実行できます。

  • Spark in MapReduce (SIMR)− MapReduceのSparkは、スタンドアロン展開に加えて、sparkジョブを起動するために使用されます。SIMRを使用すると、ユーザーはSparkを起動し、管理アクセスなしでそのシェルを使用できます。

Sparkのコンポーネント

次の図は、Sparkのさまざまなコンポーネントを示しています。

Apache Spark Core

Spark Coreは、他のすべての機能が構築されているSparkプラットフォームの基盤となる一般的な実行エンジンです。インメモリコンピューティングと外部ストレージシステムのデータセットの参照を提供します。

Spark SQL

Spark SQLは、SchemaRDDと呼ばれる新しいデータ抽象化を導入するSpark Core上のコンポーネントであり、構造化データと半構造化データのサポートを提供します。

Sparkストリーミング

Spark Streamingは、SparkCoreの高速スケジューリング機能を利用してストリーミング分析を実行します。ミニバッチにデータを取り込み、それらのデータのミニバッチに対してRDD（Resilient Distributed Datasets）変換を実行します。

MLlib（機械学習ライブラリ）

MLlibは、分散メモリベースのSparkアーキテクチャにより、Sparkより上の分散機械学習フレームワークです。ベンチマークによると、これはMLlib開発者によってAlternating Least Squares（ALS）の実装に対して行われます。Spark MLlibは、Hadoopディスクベースバージョンの9倍の速度です。Apache Mahout （MahoutがSparkインターフェイスを取得する前）。

GraphX

GraphXは、Spark上にある分散グラフ処理フレームワークです。Pregel抽象化APIを使用してユーザー定義のグラフをモデル化できるグラフ計算を表現するためのAPIを提供します。また、この抽象化のために最適化されたランタイムを提供します。

復元力のある分散データセット

復元力のある分散データセット（RDD）は、Sparkの基本的なデータ構造です。これは、オブジェクトの不変の分散コレクションです。RDDの各データセットは論理パーティションに分割され、クラスターのさまざまなノードで計算できます。RDDには、ユーザー定義クラスを含む、任意のタイプのPython、Java、またはScalaオブジェクトを含めることができます。

正式には、RDDは読み取り専用のパーティション化されたレコードのコレクションです。RDDは、安定したストレージ上のデータまたは他のRDDのいずれかに対する決定論的操作によって作成できます。RDDは、並行して操作できるフォールトトレラントな要素のコレクションです。

RDDを作成する方法は2つあります- parallelizing ドライバープログラムの既存のコレクション、または referencing a dataset 共有ファイルシステム、HDFS、HBase、またはHadoop入力形式を提供する任意のデータソースなどの外部ストレージシステム内。

Sparkは、RDDの概念を利用して、より高速で効率的なMapReduce操作を実現します。まず、MapReduce操作がどのように行われるのか、そしてなぜそれらがそれほど効率的でないのかについて説明しましょう。

MapReduceでのデータ共有が遅い

MapReduceは、クラスター上で並列分散アルゴリズムを使用して大規模なデータセットを処理および生成するために広く採用されています。これにより、ユーザーは、作業の分散やフォールトトレランスを気にすることなく、一連の高レベルの演算子を使用して並列計算を記述できます。

残念ながら、現在のほとんどのフレームワークでは、計算間（例：2つのMapReduceジョブ間）でデータを再利用する唯一の方法は、外部の安定したストレージシステム（例：HDFS）にデータを書き込むことです。このフレームワークは、クラスターの計算リソースにアクセスするための多数の抽象化を提供しますが、ユーザーはさらに多くのことを望んでいます。

どちらも Iterative そして Interactiveアプリケーションでは、並列ジョブ間でのより高速なデータ共有が必要です。MapReduceでのデータ共有は、replication、 serialization、および disk IO。ストレージシステムに関しては、ほとんどのHadoopアプリケーションで、90％以上の時間をHDFSの読み取り/書き込み操作に費やしています。

MapReduceの反復操作

多段階アプリケーションで複数の計算にわたって中間結果を再利用します。次の図は、MapReduceで反復操作を実行しているときに、現在のフレームワークがどのように機能するかを説明しています。これにより、データレプリケーション、ディスクI / O、およびシリアル化によりかなりのオーバーヘッドが発生し、システムの速度が低下します。

MapReduceのインタラクティブ操作

ユーザーは、データの同じサブセットに対してアドホッククエリを実行します。各クエリは、安定したストレージでディスクI / Oを実行します。これは、アプリケーションの実行時間を支配する可能性があります。

次の図は、MapReduceでインタラクティブクエリを実行しているときに現在のフレームワークがどのように機能するかを説明しています。

SparkRDDを使用したデータ共有

MapReduceでのデータ共有は、 replication、 serialization、および disk IO。ほとんどのHadoopアプリケーションは、90％以上の時間をHDFSの読み取り/書き込み操作に費やしています。

この問題を認識して、研究者はApacheSparkと呼ばれる特殊なフレームワークを開発しました。スパークの重要なアイデアはR弾力性 D配布 Dアタセット（RDD）; インメモリ処理の計算をサポートします。つまり、メモリの状態をジョブ間でオブジェクトとして保存し、オブジェクトはそれらのジョブ間で共有可能です。メモリ内のデータ共有は、ネットワークやディスクよりも10倍から100倍高速です。

ここで、SparkRDDで反復的でインタラクティブな操作がどのように行われるかを調べてみましょう。

SparkRDDでの反復操作

以下の図は、SparkRDDでの反復操作を示しています。中間結果を安定ストレージ（ディスク）ではなく分散メモリに保存し、システムを高速化します。

Note −分散メモリ（RAM）が中間結果（ジョブの状態）を保存するのに十分な場合、それらの結果をディスクに保存します

SparkRDDでのインタラクティブな操作

この図は、SparkRDDでのインタラクティブな操作を示しています。同じデータセットに対して異なるクエリが繰り返し実行される場合、この特定のデータをメモリに保持して、実行時間を短縮できます。

デフォルトでは、変換された各RDDは、アクションを実行するたびに再計算される場合があります。ただし、persistメモリ内のRDD。この場合、Sparkは、次回クエリを実行したときに、要素をクラスター上に保持して、はるかに高速なアクセスを実現します。ディスク上にRDDを永続化すること、または複数のノード間で複製することもサポートされています。

SparkはHadoopのサブプロジェクトです。したがって、SparkをLinuxベースのシステムにインストールすることをお勧めします。次の手順は、ApacheSparkをインストールする方法を示しています。

ステップ1：Javaインストールの確認

Javaのインストールは、Sparkをインストールする際に必須のことの1つです。次のコマンドを試して、JAVAのバージョンを確認してください。

$java -version

Javaがすでにシステムにインストールされている場合は、次の応答が表示されます。

java version "1.7.0_71"
Java(TM) SE Runtime Environment (build 1.7.0_71-b13)
Java HotSpot(TM) Client VM (build 25.0-b02, mixed mode)

システムにJavaがインストールされていない場合は、次の手順に進む前にJavaをインストールしてください。

ステップ2：Scalaのインストールを確認する

Sparkを実装するにはScala言語が必要です。それでは、次のコマンドを使用してScalaのインストールを確認しましょう。

$scala -version

Scalaがすでにシステムにインストールされている場合は、次の応答が表示されます-

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

システムにScalaがインストールされていない場合は、Scalaのインストールの次のステップに進みます。

ステップ3：Scalaをダウンロードする

次のリンクにアクセスして、Scalaの最新バージョンをダウンロードしてください。Scalaをダウンロードしてください。このチュートリアルでは、scala-2.11.6バージョンを使用しています。ダウンロード後、ダウンロードフォルダにScalatarファイルがあります。

ステップ4：Scalaをインストールする

Scalaをインストールするには、以下の手順に従ってください。

Scalatarファイルを抽出します

Scala tarファイルを抽出するには、次のコマンドを入力します。

$ tar xvf scala-2.11.6.tgz

Scalaソフトウェアファイルを移動する

Scalaソフトウェアファイルをそれぞれのディレクトリに移動するには、次のコマンドを使用します (/usr/local/scala)。

$ su –
Password:
# cd /home/Hadoop/Downloads/
# mv scala-2.11.6 /usr/local/scala
# exit

ScalaのPATHを設定する

ScalaのPATHを設定するには、次のコマンドを使用します。

$ export PATH = $PATH:/usr/local/scala/bin

Scalaのインストールの確認

インストール後、確認することをお勧めします。Scalaのインストールを確認するには、次のコマンドを使用します。

$scala -version

Scalaがすでにシステムにインストールされている場合は、次の応答が表示されます-

Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

ステップ5：ApacheSparkをダウンロードする

次のリンクにアクセスして、Sparkの最新バージョンをダウンロードします。Sparkのダウンロード。このチュートリアルでは、spark-1.3.1-bin-hadoop2.6バージョン。ダウンロード後、ダウンロードフォルダにSparktarファイルがあります。

ステップ6：Sparkのインストール

Sparkをインストールするには、以下の手順に従ってください。

スパークタールの抽出

Sparktarファイルを抽出するための次のコマンド。

$ tar xvf spark-1.3.1-bin-hadoop2.6.tgz

Sparkソフトウェアファイルの移動

Sparkソフトウェアファイルをそれぞれのディレクトリに移動するための次のコマンド (/usr/local/spark)。

$ su –
Password:
# cd /home/Hadoop/Downloads/
# mv spark-1.3.1-bin-hadoop2.6 /usr/local/spark
# exit

Sparkの環境をセットアップする

〜に次の行を追加します/.bashrcファイル。これは、sparkソフトウェアファイルが配置されている場所をPATH変数に追加することを意味します。

export PATH = $PATH:/usr/local/spark/bin

〜/ .bashrcファイルを取得するには、次のコマンドを使用します。

$ source ~/.bashrc

ステップ7：Sparkのインストールを確認する

Sparkシェルを開くには次のコマンドを記述します。

$spark-shell

Sparkが正常にインストールされると、次の出力が表示されます。

Spark assembly has been built with Hive, including Datanucleus jars on classpath
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
15/06/04 15:25:22 INFO SecurityManager: Changing view acls to: hadoop
15/06/04 15:25:22 INFO SecurityManager: Changing modify acls to: hadoop
disabled; ui acls disabled; users with view permissions: Set(hadoop); users with modify permissions: Set(hadoop)
15/06/04 15:25:22 INFO HttpServer: Starting HTTP Server
15/06/04 15:25:23 INFO Utils: Successfully started service 'HTTP class server' on port 43292.
Welcome to
    ____             __
   / __/__ ___ _____/ /__
   _\ \/ _ \/ _ `/ __/ '_/
   /___/ .__/\_,_/_/ /_/\_\ version 1.4.0
      /_/
Using Scala version 2.10.4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.7.0_71)
Type in expressions to have them evaluated.
Spark context available as sc
scala>

Sparkは、SparkSQLと呼ばれる構造化データ処理用のプログラミングモジュールを導入しています。DataFrameと呼ばれるプログラミングの抽象化を提供し、分散SQLクエリエンジンとして機能できます。

SparkSQLの機能

以下はSparkSQLの機能です-

  • Integrated−SQLクエリとSparkプログラムをシームレスに組み合わせる。Spark SQLを使用すると、Python、Scala、Javaの統合APIを使用して、構造化データをSparkの分散データセット（RDD）としてクエリできます。この緊密な統合により、複雑な分析アルゴリズムと一緒にSQLクエリを簡単に実行できます。

  • Unified Data Access−さまざまなソースからデータをロードしてクエリします。Schema-RDDは、Apache Hiveテーブル、寄木細工のファイル、JSONファイルなどの構造化データを効率的に操作するための単一のインターフェイスを提供します。

  • Hive Compatibility−既存のウェアハウスで変更されていないHiveクエリを実行します。Spark SQLはHiveフロントエンドとMetaStoreを再利用し、既存のHiveデータ、クエリ、およびUDFとの完全な互換性を提供します。Hiveと一緒にインストールするだけです。

  • Standard Connectivity−JDBCまたはODBCを介して接続します。Spark SQLには、業界標準のJDBCおよびODBC接続を備えたサーバーモードが含まれています。

  • Scalability−インタラクティブクエリと長いクエリの両方に同じエンジンを使用します。Spark SQLは、RDDモデルを利用してクエリ中のフォールトトレランスをサポートし、大規模なジョブにも拡張できるようにします。履歴データに別のエンジンを使用することを心配する必要はありません。

SparkSQLアーキテクチャ

次の図は、SparkSQLのアーキテクチャを説明しています-

このアーキテクチャには、言語API、スキーマRDD、データソースの3つのレイヤーが含まれています。

  • Language API−Sparkはさまざまな言語およびSparkSQLと互換性があります。また、これらの言語-API（python、scala、java、HiveQL）でもサポートされています。

  • Schema RDD− Spark Coreは、RDDと呼ばれる特別なデータ構造で設計されています。通常、Spark SQLはスキーマ、テーブル、およびレコードで機能します。したがって、スキーマRDDを一時テーブルとして使用できます。このスキーマRDDをデータフレームと呼ぶことができます。

  • Data Sources−通常、spark-coreのデータソースはテキストファイル、Avroファイルなどです。ただし、SparkSQLのデータソースは異なります。それらは、Parquetファイル、JSONドキュメント、HIVEテーブル、およびCassandraデータベースです。

これらについては、次の章で詳しく説明します。

DataFrameはデータの分散コレクションであり、名前付きの列に編成されています。概念的には、優れた最適化手法を備えたリレーショナルテーブルと同等です。

DataFrameは、Hiveテーブル、構造化データファイル、外部データベース、既存のRDDなどのさまざまなソースの配列から構築できます。このAPIは、最新のビッグデータおよびデータサイエンスアプリケーション向けに設計されました。DataFrame in R Programming そして Pandas in Python。

DataFrameの機能

これがDataFrameのいくつかの特徴的な機能のセットです-

  • 単一ノードクラスターから大規模クラスターでキロバイトからペタバイトのサイズのデータ​​を処理する機能。

  • さまざまなデータ形式（Avro、csv、Elastic Search、Cassandra）とストレージシステム（HDFS、HIVEテーブル、mysqlなど）をサポートします。

  • Spark SQL Catalystオプティマイザー（ツリー変換フレームワーク）による最先端の最適化とコード生成。

  • Spark-Coreを介して、すべてのビッグデータツールおよびフレームワークと簡単に統合できます。

  • Python、Java、Scala、およびRプログラミング用のAPIを提供します。

SQLContext

SQLContextはクラスであり、SparkSQLの機能を初期化するために使用されます。SQLContextクラスオブジェクトを初期化するには、SparkContextクラスオブジェクト（sc）が必要です。

次のコマンドは、spark-shellを介してSparkContextを初期化するために使用されます。

$ spark-shell

デフォルトでは、SparkContextオブジェクトは次の名前で初期化されます sc スパークシェルが開始したとき。

次のコマンドを使用して、SQLContextを作成します。

scala> val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

例

名前の付いたJSONファイルの従業員レコードの例を考えてみましょう employee.json。次のコマンドを使用して、DataFrame（df）を作成し、という名前のJSONドキュメントを読み取ります。employee.json 以下の内容で。

employee.json −このファイルを現在のディレクトリに配置します scala> ポインタがあります。

{
   {"id" : "1201", "name" : "satish", "age" : "25"}
   {"id" : "1202", "name" : "krishna", "age" : "28"}
   {"id" : "1203", "name" : "amith", "age" : "39"}
   {"id" : "1204", "name" : "javed", "age" : "23"}
   {"id" : "1205", "name" : "prudvi", "age" : "23"}
}

DataFrame操作

DataFrameは、構造化データ操作用のドメイン固有言語を提供します。ここでは、DataFrameを使用した構造化データ処理の基本的な例をいくつか紹介します。

以下の手順に従って、DataFrame操作を実行します-

JSONドキュメントを読む

まず、JSONドキュメントを読む必要があります。これに基づいて、（dfs）という名前のDataFrameを生成します。

次のコマンドを使用して、という名前のJSONドキュメントを読み取ります employee.json。データは、フィールド-id、name、およびageを持つテーブルとして表示されます。

scala> val dfs = sqlContext.read.json("employee.json")

Output −フィールド名はから自動的に取得されます employee.json。

dfs: org.apache.spark.sql.DataFrame = [age: string, id: string, name: string]

データを表示する

DataFrameのデータを表示する場合は、次のコマンドを使用します。

scala> dfs.show()

Output −従業員データを表形式で表示できます。

<console>:22, took 0.052610 s
+----+------+--------+
|age | id   |  name  |
+----+------+--------+
| 25 | 1201 | satish |
| 28 | 1202 | krishna|
| 39 | 1203 | amith  |
| 23 | 1204 | javed  |
| 23 | 1205 | prudvi |
+----+------+--------+

printSchemaメソッドを使用する

DataFrameの構造（スキーマ）を確認する場合は、次のコマンドを使用します。

scala> dfs.printSchema()

Output

root
   |-- age: string (nullable = true)
   |-- id: string (nullable = true)
   |-- name: string (nullable = true)

選択方法を使用する

次のコマンドを使用してフェッチします name-DataFrameの3つの列のうちの列。

scala> dfs.select("name").show()

Output −の値を確認できます name カラム。

<console>:22, took 0.044023 s
+--------+
|  name  |
+--------+
| satish |
| krishna|
| amith  |
| javed  |
| prudvi |
+--------+

年齢フィルターを使用する

次のコマンドを使用して、23歳を超える（23歳を超える）従業員を検索します。

scala> dfs.filter(dfs("age") > 23).show()

Output

<console>:22, took 0.078670 s
+----+------+--------+
|age | id   | name   |
+----+------+--------+
| 25 | 1201 | satish |
| 28 | 1202 | krishna|
| 39 | 1203 | amith  |
+----+------+--------+

groupByメソッドを使用する

次のコマンドを使用して、同じ年齢の従業員の数をカウントします。

scala> dfs.groupBy("age").count().show()

Output −2人の従業員は23歳です。

<console>:22, took 5.196091 s
+----+-----+
|age |count|
+----+-----+
| 23 |  2  |
| 25 |  1  |
| 28 |  1  |
| 39 |  1  |
+----+-----+

プログラムによるSQLクエリの実行

SQLContextを使用すると、アプリケーションはSQL関数の実行中にプログラムでSQLクエリを実行し、結果をDataFrameとして返します。

一般に、バックグラウンドで、SparkSQLは既存のRDDをDataFrameに変換するための2つの異なる方法をサポートします-

シニアいいえ 方法と説明
1 リフレクションを使用したスキーマの推測

このメソッドは、リフレクションを使用して、特定のタイプのオブジェクトを含むRDDのスキーマを生成します。
2 プログラムによるスキーマの指定

DataFrameを作成するための2番目の方法は、スキーマを構築して既存のRDDに適用できるようにするプログラムインターフェイスを使用することです。

DataFrameインターフェースにより、さまざまなデータソースがSparkSQLで機能することができます。これは一時テーブルであり、通常のRDDとして操作できます。DataFrameをテーブルとして登録すると、そのデータに対してSQLクエリを実行できます。

この章では、さまざまなSparkデータソースを使用してデータをロードおよび保存するための一般的な方法について説明します。その後、組み込みデータソースで使用できる特定のオプションについて詳しく説明します。

SparkSQLで利用できるデータソースにはさまざまな種類があり、その一部を以下に示します。

シニアいいえ データソース
1 JSONデータセット

Spark SQLは、JSONデータセットのスキーマを自動的にキャプチャし、それをDataFrameとしてロードできます。
2 ハイブテーブル

Hiveは、SQLContextを継承するHiveContextとしてSparkライブラリにバンドルされています。
3 寄木細工のファイル

Parquetは列形式であり、多くのデータ処理システムでサポートされています。
ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved