--Faites le tutoriel Spark sur la page suivante
export SPARK_HOME='/usr/local/bin/spark-2.2.0-bin-hadoop2.7'
Si vous sélectionnez without-hadoop lorsque vous l'installez, vous devez insérer hadoop et définir HADOOP_HOME.
Lors de l'utilisation de PySpark
pip pyspark
Je l'ai mis dans quelque chose. Si la version est différente de Spark, cela peut ne pas fonctionner correctement, donc si vous souhaitez spécifier la version
pip pyspark-2.2.1
Précisez en faisant quelque chose comme ça.
La structure des répertoires ressemble à ceci. Le contenu sera expliqué ci-dessous.
$ tree
.
├── SimpleApp.scala
├── build.sbt
├── input.txt
└── run.sh
SimpleApp.scala est presque identique au tutoriel. Seule l'entrée a été légèrement modifiée. Lisez le fichier texte et comptez le nombre de "a" et "p" inclus.
SimpleApp.scala
/* SimpleApp.scala */
import org.apache.spark.sql.SparkSession
object SimpleApp {
def main(args: Array[String]) {
val logFile = "input.txt" // Should be some file on your system
val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
val logData = spark.read.textFile(logFile).cache()
val numAs = logData.filter(line => line.contains("a")).count()
val numPs = logData.filter(line => line.contains("p")).count()
println(s"Lines with a: $numAs, Lines with p: $numPs")
spark.stop()
}
}
Phrases appropriées dans input.txt
this is a pen
this is an apple
apple pen
pen pen
sbt package
Générera un fichier jar sous cible basé sur le contenu de build.sbt. (La cible est également créée arbitrairement à ce moment) Exemple de build.sbt
build.sbt
name := "simple"
version := "1.0"
scalaVersion := "2.11.12"
libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.0.0"
Dans ce cas, le fichier jar généré sera target / scala-2.11 / simple_2.11-1.0.jar.
run.sh est un script pour l'exécution.
run.sh
spark-submit \
--class SimpleApp \
--master local[4] \
--conf spark.driver.host=localhost \
target/scala-2.11/simple_2.11-1.0.jar
Quand je lance sh run.sh, les résultats sont échelonnés,
Lines with a: 3, Lines with p: 4
S'il est inclus, c'est correct.
** Parmi les entrées, 3 lignes contiennent a et 4 lignes contiennent p **.
La partie spark.driver.host = localhost n'était pas incluse dans le tutoriel, mais si vous n'écrivez pas ceci, ce sera votre environnement
Error
Exception in thread "main" java.lang.AssertionError: assertion failed: Expected hostname
Je suis sorti, alors je l'ai ajouté.
C'est un peu plus facile pour Python. La structure des fichiers est la suivante
$ tree
.
├── SimpleApp.py
├── input.txt
└── run.sh
SimpleApp.py
from pyspark.sql import SparkSession
logFile = "input.txt" # Should be some file on your system
spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
logData = spark.read.text(logFile).cache()
numAs = logData.filter(logData.value.contains('a')).count()
numPs = logData.filter(logData.value.contains('p')).count()
print("Lines with a: %i, lines with p: %i" % (numAs, numPs))
spark.stop()
run.sh
spark-submit \
--master local[4] \
--conf spark.driver.host=localhost \
SimpleApp.py
input.txt est le même. Si Pypark est inclus dans pip, vous n'avez pas à exécuter le script
python SimpleApp.py
Il semble qu'il puisse être exécuté tout seul.
Cette fois, j'exécute Spark avec Spark Session. Vous pouvez faire de même avec SparkContext. Si vous ne comprenez pas la différence entre Spark Context et Spark Session, il semble que Spark Session contient Spark Context et apparaisse après Spark Context.
spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
Créez une instance Spark en spécifiant appName avec.
logData = spark.read.text(logFile).cache()
Lisez le fichier avec. cache () semble être un paramètre qui persiste en mémoire. (Cela a fonctionné sans ce code) Cette fois, nous lisons un fichier texte, mais nous pouvons également lire des données structurées telles que des fichiers csv et des tables sql.
Les données lues sont parallélisées et calculées par «rdd». Diverses fonctions telles que map et fileter sont préparées dans rdd.
--Référence: https://dev.classmethod.jp/articles/apache-spark_rdd_investigation/
filter est une méthode de filtrage littéral. Cette fois, je lis le fichier ligne par ligne et j'utilise count () pour compter combien contiennent une chaîne particulière.
numAs = logData.filter(logData.value.contains('a')).count()
SparkSession semble fonctionner en tant que serveur http, alors arrêtez-le avec stop () à la fin.
spark.stop()
Recommended Posts