Spark SQL の 4 つの一般的なデータ ソースの詳細な説明

scala> val df = spark.read.load("hdfs://hadoop001:9000/namesAndAges.parquet")
df: org.apache.spark.sql.DataFrame = [年齢: bigint、名前: 文字列]
scala> df.select("name").write.save("names.parquet")

scala> val peopleDF = spark.read.format("json").load("hdfs://hadoop001:9000/people.json")
peopleDF: org.apache.spark.sql.DataFrame = [年齢: bigint, 名前: 文字列]
scala> peopleDF.write.format("parquet").save("hdfs://hadoop001:9000/namesAndAges.parquet")
スカラ>

val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop001:9000/namesAndAges.parquet`")
sqlDF.表示()

// 最も一般的なタイプのエンコーダーは、spark.implicits をインポートすることで自動的に提供されます。
spark.implicits._ をインポートします。
val peopleDF = spark.read.json("examples/src/main/resources/people.json")
// DataFramesはスキーマ情報を維持しながらParquetファイルとして保存できます
peopleDF.write.parquet("hdfs://hadoop001:9000/people.parquet")
// 上記で作成した parquet ファイルを読み込む
// Parquetファイルは自己記述型なのでスキーマは保持されます
// Parquet ファイルをロードした結果も DataFrame になります
parquetFileDF を spark.read.parquet("hdfs://hadoop001:9000/people.parquet") に変更します。
// Parquet ファイルは一時ビューの作成にも使用でき、その後 SQL ステートメントで使用できます。
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("parquetFile から name を選択し、age を 13 から 19 の間で指定します")
namesDF.map(attributes => "名前: " + 属性(0)).show()
// +------------+
// | 値|
// +------------+
// |名前: ジャスティン|
// +------------+

パス
└──に
└── テーブル
├── 性別=男性
│ ├──……
│ │
│ ├── 国=米国
│ │ └── データ.parquet
│ ├── 国=CN
│ │ └── データ.parquet
│ └──……
└── 性別=女性
├──……
│
├── 国=米国
│ └── データ.parquet
├── 国=CN
│ └── データ.parquet
└──……

根
|-- 名前: 文字列 (nullable = true)
|-- age: long (nullable = true)
|-- 性別: 文字列 (nullable = true)
|-- 国: 文字列 (nullable = true)

spark.sql.sources.partitionColumnTypeInference が有効

// この例では、前の例の sqlContext が使用されます。
// これは RDD を暗黙的に DataFrame に変換するために使用されます。
spark.implicits._ をインポートします。
// パーティションディレクトリに保存される単純な DataFrame を作成します
val df1 = sc.makeRDD(1 から 5).map(i => (i, i * 2)).toDF("single", "double")
df1.write.parquet("hdfs://hadoop001:9000/data/test_table/key=1")
// 新しいパーティションディレクトリに別の DataFrame を作成します。
// 新しい列を追加し、既存の列を削除します
val df2 = sc.makeRDD(6 から 10).map(i => (i, i * 3)).toDF("シングル", "トリプル")
df2.write.parquet("hdfs://hadoop001:9000/data/test_table/key=2")
// パーティションテーブルを読み込む
val df3 = spark.read.option("mergeSchema", "true").parquet("hdfs://hadoop001:9000/data/test_table")
df3.printSchema()
// 最終的なスキーマは、Parquet ファイル内の 3 つの列すべてで構成されます
// パーティション ディレクトリ パスにパーティション列が表示されます。
//根
// |-- 単一: int (nullable = true)
// |-- double: int (nullable = true)
// |-- トリプル: int (nullable = true)
// |-- キー: int (nullable = true)

java.io.File をインポートする
org.apache.spark.sql.Row をインポートします。
org.apache.spark.sql.SparkSession をインポートします。
ケースクラス Record(キー: Int、値: String)
// warehouseLocation は管理対象データベースとテーブルのデフォルトの場所を指します
val warehouseLocation = 新しいファイル("spark-warehouse").getAbsolutePath
val スパーク = SparkSession
。ビルダー（）
.appName("Spark Hive の例")
.config("spark.sql.warehouse.dir", 倉庫の場所)
.enableHiveSupport()
.getOrCreate()
spark.implicits._ をインポートします。
spark.sqlをインポートする
sql("テーブルが存在しない場合は作成します src (キー INT、値 STRING)")
sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' を TABLE src にロード")
// クエリはHiveQLで表現されます
sql("SELECT * FROM src").show()
// +---+-------+
// |キー| 値|
// +---+-------+
//|238|val_238|
//| 86| val_86|
//|311|val_311|
// ...
// 集計クエリもサポートされています。
sql("SELECT COUNT(*) FROM src").show()
// +--------+
// |カウント(1)|
// +--------+
// | 500 |
// +--------+
// SQL クエリの結果自体は DataFrame であり、すべての通常の関数をサポートします。
val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key")
// DataFrame 内の項目は Row 型なので、序数で各列にアクセスできます。
val stringsDS = sqlDF.map {
case Row(キー: Int, 値: String) => s"キー: $key, 値: $value"
}
文字列DS.show()
//+--------------------+
// | 値|
// +--------------------+
// |キー: 0、値: val_0|
// |キー: 0、値: val_0|
// |キー: 0、値: val_0|
// ...
// DataFrames を使用して、SparkSession 内に一時ビューを作成することもできます。
val recordsDF = spark.createDataFrame((1 から 100).map(i => Record(i, s"val_$i")))
recordsDF.createOrReplaceTempView("レコード")
// クエリは、DataFrame データを Hive に保存されているデータと結合できます。
sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()
// +---+------+---+------+
// |キー| 値|キー| 値|
// +---+------+---+------+
// | 2| val_2| 2| val_2|
// | 4| val_4| 4| val_4|
// | 5| val_5| 5| val_5|
// ...

spark.sql.warehouse.dir=

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar

{"名前":"マイケル"}
{"名前":"アンディ", "年齢":30}
{"名前":"ジャスティン", "年齢":19}
// プリミティブ型（Int、Stringなど）と積型（ケースクラス）のエンコーダは
// データセットを作成するときにこれをインポートすることでサポートされます。
spark.implicits._ をインポートします。
// JSON データセットはパスによって指し示されます。
// パスは単一のテキストファイルまたはテキストファイルを格納するディレクトリのいずれかになります
val パス = "examples/src/main/resources/people.json"
val peopleDF = spark.read.json(パス)
// 推論されたスキーマはprintSchema()メソッドを使用して視覚化できます
peopleDF.printSchema()
//根
// |-- age: long (nullable = true)
// |-- name: 文字列 (nullable = true)
// DataFrameを使用して一時ビューを作成します
peopleDF.createOrReplaceTempView("人")
// SQL文はSparkが提供するSQLメソッドを使って実行できる
val teenagerNamesDF = spark.sql("年齢が13から19の間であるpeopleからnameを選択")
ティーンエイジャー名DF.show()
// +------+
// | 名前|
// +------+
// |ジャスティン|
// +------+
// あるいは、JSONデータセットのDataFrameを作成することもできます。
// 文字列ごとに1つのJSONオブジェクトを格納するDataset[String]
val otherPeopleDataset = spark.createDataset(
"""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
val otherPeople = spark.read.json(otherPeopleDataset)
他のユーザー.表示()
// +--------------+----+
// | 住所|名前|
// +--------------+----+
// |[オハイオ州コロンバス]| 陰|
// +--------------+----+

$ bin/spark-shell --master spark://hadoop001:7077 --jars mysql-connector-java-5.1.27-bin.jar
// 注: JDBC の読み込みと保存は、load/save メソッドまたは jdbc メソッドのいずれかを使用して実行できます。
// JDBCソースからデータをロードする
val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://hadoop001:3306/rdd").option("dbtable", "rddtable").option("user", "root").option("password", "hive").load()
val connectionProperties = 新しい Properties()
connectionProperties.put("user", "root")
connectionProperties.put("パスワード", "ハイブ")
val jdbcDF2 = spark.read
.jdbc("jdbc:mysql://hadoop001:3306/rdd", "rddtable", 接続プロパティ)
// データをJDBCソースに保存する
jdbcDF.書き込み
.format("jdbc")
.option("url", "jdbc:mysql://hadoop001:3306/rdd")
.option("dbtable", "rddtable2")
.option("ユーザー", "ルート")
.option("パスワード", "ハイブ")
。保存（）
jdbcDF2.書き込み
.jdbc("jdbc:mysql://hadoop001:3306/mysql", "db", 接続プロパティ)
// 書き込み時にテーブル列のデータ型を指定する
jdbcDF.書き込み
.option("createTableColumnTypes", "名前 CHAR(64), コメント VARCHAR(1024)")
.jdbc("jdbc:mysql://hadoop001:3306/mysql", "db", 接続プロパティ)