Создание потока из текстового файла в Pyspark - PullRequest
Новая
       31

Создание потока из текстового файла в Pyspark

0 голосов
/ 21 октября 2018

При попытке создать поток из текстового файла в Pyspark появляется следующая ошибка:

TypeError: unbound method textFileStream() must be called with StreamingContext instance as first argument (got str instance instead)

Я не хочу использовать SparkContext, потому чтоЯ получаю еще одну ошибку, поэтому для ее устранения я должен использовать SparkSession.

Мой код: 

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.mllib.stat import Statistics
if __name__ == "__main__":

   spark = SparkSession.builder.appName("CrossCorrelation").getOrCreate()
   ssc = StreamingContext(spark.sparkContext, 5)
   input_path1 = sys.argv[1]
   input_path2 = sys.argv[2]
   ds1 = ssc.textFileStream(input_path1)
   lines1 = ds1.map(lambda x1: x1[1])
   windowedds1 = lines1.flatMap(lambda line1: line1.strip().split("\n")).map(lambda strelem1: float(strelem1)).window(5,10)
   ds2 = ssc.textFileStream(input_path2)
   lines2 = ds2.map(lambda x2: x2[1])
   windowedds2 = lines2.flatMap(lambda line2: line2.strip().split("\n")).map(lambda strelem2: float(strelem2)).window(5,10)







   result = Statistics.corr(windowedds1,windowedds2, method="pearson")
   if result > 0.7:
    print("ds1 and ds2 are correlated!!!")


   spark.stop()

Спасибо!

1 Ответ

0 голосов
/ 21 октября 2018

Вы должны сначала создать объект streamingcontext, а затем использовать его для вызова textFileStream.

spark = 
SparkSession.builder.appName("CrossCorrelation").getOrCreate()
ssc = StreamingContext(spark.sparkContext, 1)
ds = ssc.textFileStream(input_path)
...