재현 가능한 좋은 Apache Spark 예제를 만드는 방법

재현 가능한 좋은 Apache Spark 예제를 만드는 방법

저는 pyspark 및 spark-dataframe 태그를 사용하여 몇 가지 질문을 읽는 데 상당한 시간을 보냈습니다. 포스터가 질문을 제대로 이해하기에 충분한 정보를 제공하지 못한다는 사실을 자주 알게 되었습니다.저는 보통 MCVE를 올려달라고 요청하지만 때로는 샘플 입출력 데이터를 보여주도록 요청하는 것은 이를 뽑는 것과 같습니다.

아마도 문제의 일부는 사람들이 스파크 데이터 프레임을 위한 MCVE를 쉽게 만드는 방법을 모른다는 것입니다.이 팬더 질문의 스파크 데이터 프레임 버전을 연동할 수 있는 가이드로 제공하는 것이 유용할 것 같습니다.

그렇다면 어떻게 하면 좋은, 재현 가능한 예를 만들 수 있을까요?

쉽게 다시 만들 수 있는 작은 샘플 데이터를 제공합니다.

최소한 포스터는 데이터 프레임과 코드에 몇 개의 행과 열을 제공하여 쉽게 만들 수 있도록 해야 합니다.쉽게 말하자면, 자르고 붙여넣기를 의미합니다.문제를 입증할 수 있도록 가능한 작게 합니다.

---

저는 다음과 같은 데이터 프레임을 가지고 있습니다.

+-----+---+-----+----------+
|index|  X|label|      date|
+-----+---+-----+----------+
|    1|  1|    A|2017-01-01|
|    2|  3|    B|2017-01-02|
|    3|  5|    A|2017-01-03|
|    4|  7|    B|2017-01-04|
+-----+---+-----+----------+

다음 코드로 만들 수 있습니다.

df = sqlCtx.createDataFrame(
    [
        (1, 1, 'A', '2017-01-01'),
        (2, 3, 'B', '2017-01-02'),
        (3, 5, 'A', '2017-01-03'),
        (4, 7, 'B', '2017-01-04')
    ],
    ('index', 'X', 'label', 'date')
)

---

원하는 출력을 표시합니다.

구체적인 질문을 하고 원하는 결과물을 보여주십시오.

---

새 열을 만들려면 어떻게 해야 합니까? 'is_divisible' 가치가 있는. 'yes' 만약 그 달의 날에 'date' + 7일은 열에 있는 값으로 나눕니다.'X', 그리고 'no' 그렇지 않다면요?

원하는 출력:

+-----+---+-----+----------+------------+
|index|  X|label|      date|is_divisible|
+-----+---+-----+----------+------------+
|    1|  1|    A|2017-01-01|         yes|
|    2|  3|    B|2017-01-02|         yes|
|    3|  5|    A|2017-01-03|         yes|
|    4|  7|    B|2017-01-04|          no|
+-----+---+-----+----------+------------+

---

출력을 얻는 방법을 설명합니다.

원하는 결과물을 얻는 방법을 자세히 설명합니다.계산 예제를 보여 주는 데 도움이 됩니다.

---

예를 들어 1행에서 X = 1 및 날짜 = 2017-01-01입니다.현재까지 7일을 더하면 2017-01-08이 됩니다.그 달의 요일은 8이고 8은 1로 나눌 수 있기 때문에 '네'라고 대답합니다.

마찬가지로 마지막 행 X = 7 및 날짜 = 2017-01-04의 경우.날짜에 7을 더하면 11이 이달의 날로 계산됩니다.11% 7은 0이 아니기 때문에 '아니오'라고 대답합니다.

---

기존 코드를 공유합니다.

작동하지 않더라도 코드의 모든*을 포함하여 귀하가 수행하거나 시도한 작업을 보여주십시오.어디서 막히게 되는지 알려주시고 오류가 발생하면 오류 메시지를 포함하여 주시기 바랍니다.

(*스파크 컨텍스트 생성을 위해 코드를 생략할 수 있지만, 모든 가져오기를 포함해야 합니다.)

---

새 열을 추가하는 방법을 알고 있습니다. date 7일을 더 추가했는데 이달의 날을 정수로 구하는 데 어려움을 겪고 있습니다.

from pyspark.sql import functions as f
df.withColumn("next_week", f.date_add("date", 7))

---

버전, 가져오기 및 구문 강조 표시 사용

• 데저트넛이 쓴 이 답변의 자세한 내용.

---

성능 튜닝 게시물의 경우 실행 계획을 포함합니다.

• 알퍼트가 쓴 이 답변의 자세한 내용. 터커.
• 컨텍스트에 표준화된 이름을 사용하는 데 도움이 됩니다.

---

스파크 출력 파일 구문 분석

• MaxU는 이 답변에서 Spark 출력 파일을 DataFrame으로 파싱하는 데 도움이 되는 유용한 코드를 제공했습니다.

---

다른 노트들.

• 질문하는 방법과 최소, 완료 및 검증 가능한 예제를 만드는 방법을 먼저 읽어 보십시오.
• 위에 링크된 이 질문에 대한 다른 답변을 읽어 보십시오.
• 제목이 좋고 설명력이 있습니다.
• 공손하게 하세요.SO에 계신 분들은 자원봉사자니까 예쁘게 물어보세요.

성능튜닝

성능 튜닝과 관련된 질문일 경우 다음 정보를 포함하여 주시기 바랍니다.

실행계획

확장 실행 계획을 포함하는 것이 가장 좋습니다.Python의 경우:

df.explain(True) 

스칼라에서:

df.explain(true)

또는 통계와 함께 확장 실행 계획을 세울 수 있습니다.Python의 경우:

print(df._jdf.queryExecution().stringWithStats())

스칼라:

df.queryExecution.stringWithStats

모드 및 클러스터 정보

• mode-local,client, '군집'.
• 클러스터 관리자(해당하는 경우) - 없음(로컬 모드), 독립 실행형, YRAN, Mesos, Kubernetes.
• 기본 구성 정보(코어 수, 실행기 메모리)

타이밍정보

속도가 느리다는 것은 상대적인 현상입니다. 특히 distrib되지 않은 애플리케이션을 포팅하거나 대기 시간이 짧을 것으로 예상되는 경우 더욱 그렇습니다.다양한 작업 및 스테이지에 대한 정확한 타이밍은 Spark UI에서 가져올 수 있습니다(sc.uiWebUrl)jobs아니면 스파크 REST UI.

컨텍스트에 표준화된 이름 사용

각 컨텍스트에 대해 확립된 이름을 사용하면 문제를 신속하게 재현할 수 있습니다.

• sc- …을 위하여SparkContext.
• sqlContext- …을 위하여SQLContext.
• spark- …을 위하여SparkSession.

유형 정보 제공(Scala)

강력한 유형 추론은 스칼라의 가장 유용한 기능 중 하나이지만 맥락에서 벗어난 코드를 분석하는 것을 어렵게 합니다.유형이 문맥상 명백하더라도 변수에 주석을 다는 것이 좋습니다.선호하다

val lines: RDD[String] = sc.textFile("path")
val words: RDD[String] = lines.flatMap(_.split(" "))

위에

val lines = sc.textFile("path")
val words = lines.flatMap(_.split(" "))

일반적으로 사용되는 도구는 다음과 같습니다.

• spark-shell/ 스칼라 껍질

  사용하다:t

  scala> val rdd = sc.textFile("README.md")
  rdd: org.apache.spark.rdd.RDD[String] = README.md MapPartitionsRDD[1] at textFile at <console>:24
  
  scala> :t rdd
  org.apache.spark.rdd.RDD[String]
  

• 인텔 iJ 아이디어

  사용 +

이미 제공된 내용에 대한 몇 가지 추가 제안 사항:

Spark 버전 포함

스파크는 1.x 시절처럼 빠르게 진화하고 있지는 않지만 여전히 진화하고 있습니다.작업 버전을 포함하는 것이 항상 좋습니다(특히 약간 오래된 버전을 사용하는 경우).개인적으로, 저는 항상 다음과 같이 대답을 시작합니다.

spark.version
# u'2.2.0'

아니면

sc.version
# u'2.2.0'

Python 버전을 포함하는 것도 결코 나쁘지 않은 생각입니다.

---

가져온 항목을 모두 포함합니다.

Spark SQL 및 데이터 프레임에 대한 질문이 엄격하지 않은 경우(예: 기계 학습 작업에서 데이터 프레임을 사용하고자 하는 경우) 수입품에 대해 명시적으로 설명하십시오. 이 질문을 참조하십시오.(현재 제거된) 의견에서 광범위하게 교환한 후에야 OP에 수입품이 추가되었습니다 (그리고 이러한 잘못된 수입품이 문제의 근본 원인임이 밝혀졌습니다).

이것이 왜 필요합니까?왜냐하면, 예를 들면, 이 LDA는

from pyspark.mllib.clustering import LDA

는 이 LDA와 다릅니다.

from pyspark.ml.clustering import LDA

첫 번째는 기존 RDD 기반 API(이전의 Spark MLlib)에서 나온 것이고, 두 번째는 새로운 데이터 프레임 기반 API(Spark ML)에서 나온 것입니다.

---

코드 강조 포함

자, 이것은 주관적인 것임을 고백하겠습니다: PySpark 질문에 다음과 같이 태그를 붙여서는 안 된다고 생각합니다.python 기본적으로, 문제는,python태그는 자동으로 코드 강조 표시를 제공합니다(이것이 PySpark 질문에 사용하는 사람들의 주된 이유라고 생각합니다).어쨌든, 만약 당신이 동의하고, 여전히 좋은 강조 표시된 코드를 원한다면, 관련 마크다운 지시사항을 포함하기만 하면 됩니다.

<!-- language-all: lang-python -->

당신의 게시물 어딘가에, 당신의 첫번째 코드 조각 전에.

[업데이트: 다음에 대한 자동 구문 강조 표시를 요청했습니다.pyspark그리고.sparkr실제로 구현된 태그]

이 작은 도우미 기능은 스파크 출력 파일을 DataFrame으로 구문 분석하는 데 도움이 될 수 있습니다.

PySpark:

from pyspark.sql.functions import *

def read_spark_output(file_path):
    step1 = spark.read \
             .option("header","true") \
             .option("inferSchema","true") \
             .option("delimiter","|") \
             .option("parserLib","UNIVOCITY") \
             .option("ignoreLeadingWhiteSpace","true") \
             .option("ignoreTrailingWhiteSpace","true") \
             .option("comment","+") \
             .csv("file://{}".format(file_path))
    # select not-null columns
    step2 = t.select([c for c in t.columns if not c.startswith("_")])
    # deal with 'null' string in column
    return step2.select(*[when(~col(col_name).eqNullSafe("null"), col(col_name)).alias(col_name) for col_name in step2.columns])

스칼라:

// read Spark Output Fixed width table:
def readSparkOutput(filePath: String): org.apache.spark.sql.DataFrame = {
  val step1 = spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .option("delimiter", "|")
    .option("parserLib", "UNIVOCITY")
    .option("ignoreLeadingWhiteSpace", "true")
    .option("ignoreTrailingWhiteSpace", "true")
    .option("comment", "+")
    .csv(filePath)

  val step2 = step1.select(step1.columns.filterNot(_.startsWith("_c")).map(step1(_)): _*)

  val columns = step2.columns
  columns.foldLeft(step2)((acc, c) => acc.withColumn(c, when(col(c) =!= "null", col(c))))
}

용도:

df = read_spark_output("file:///tmp/spark.out")

PS: pyspark의 경우,eqNullSafe에서 사용할 수 있습니다.spark 2.3.

언급URL : https://stackoverflow.com/questions/48427185/how-to-make-good-reproducible-apache-spark-examples