HDFSHadoopマップでインクリメンタルアップデートを処理する方法-Reduce

私はプログラムにあなたを示唆していますscalaためspark。あなたがプログラムした場合mapreduce、それだけのために有用であろうhadoopが、でプログラミングするscalaためにsparkあなたが処理できるようになるsparkだけでなく、としてhadoop。モデルSparkの欠点を処理するために開始されましたmapreduce。このトピックに関する多くのリソースを見つけることができます。それらの1つはこれです

あなたの問題に関して、私はあなたに使用することを提案しています dataframe

最初のタスクはschema、データフレーム用に作成することです。

val schema = StructType(Array(StructField("OgId", StringType),
  StructField("ItemId", StringType),
  StructField("segmentId", StringType),
  StructField("Sequence", StringType),
  StructField("Action", StringType)))

次のタスクは、2つのファイルを読み取り、上記のスキーマを使用してデータフレームを作成することです。

import org.apache.spark.sql.functions._
val textRdd1 = sparkContext.textFile("input path to file1 in hdfs")
val rowRdd1 = textRdd1.map(line => Row.fromSeq(line.split("\\|\\^\\|", -1)))
var df1 = sqlContext.createDataFrame(rowRdd1, schema)
df1 = df1.withColumn("Action", regexp_replace($"Action", "[|!|]", ""))

val textRdd2 = sparkContext.textFile("input path to file 2 in hdfs")
val rowRdd2 = textRdd2.map(line => Row.fromSeq(line.split("\\|\\^\\|", -1)))
var df2 = sqlContext.createDataFrame(rowRdd2, schema)
df2 = df2.withColumn("Action", regexp_replace($"Action", "[|!|]", ""))

の出力df1は

+----------+------+---------+--------+------+
|OgId      |ItemId|segmentId|Sequence|Action|
+----------+------+---------+--------+------+
|4295877341|136   |4        |1       |I     |
|4295877346|136   |4        |1       |I     |
|4295877341|138   |2        |1       |I     |
|4295877341|141   |4        |1       |I     |
|4295877341|143   |2        |1       |I     |
|4295877341|145   |14       |1       |I     |
+----------+------+---------+--------+------+

そしての出力df2は

+----------+------+---------+--------+------+
|OgId      |ItemId|segmentId|Sequence|Action|
+----------+------+---------+--------+------+
|4295877341|213   |4        |1       |I     |
|4295877341|215   |2        |1       |I     |
|4295877341|141   |4        |1       |I     |
|4295877341|143   |2        |1       |I     |
|4295877343|149   |14       |2       |I     |
+----------+------+---------+--------+------+

次に、要件に応じて、ifと一致する場合は削除rowsしdf1、すべてをに追加します。これらの要件は、以下のように実行できます。OgIddf2df2df1

val tempdf = df2.select("OgId").withColumnRenamed("OgId", "OgId_1")

df1 = df1.join(tempdf, df1("OgId") === tempdf("OgId_1"), "left")
df1 = df1.filter("OgId_1 is null").drop("OgId_1")
df1 = df1.union(df2)

最終出力は

+----------+------+---------+--------+------+
|OgId      |ItemId|segmentId|Sequence|Action|
+----------+------+---------+--------+------+
|4295877346|136   |4        |1       |I     |
|4295877341|213   |4        |1       |I     |
|4295877341|215   |2        |1       |I     |
|4295877341|141   |4        |1       |I     |
|4295877341|143   |2        |1       |I     |
|4295877343|149   |14       |2       |I     |
+----------+------+---------+--------+------+

この最終結果は、hdfsとして保存できます。

df1.write.format("com.databricks.spark.csv").save("output file path in hdfs")

これがお役に立てば幸いです

注：入力位置と出力位置のパスを正しく記述していることを確認してください