Styjun

What typically incentivizes a professor to change jobs to a lower ranking university?

Copycat chess is back

Is there really no realistic way for a skeleton monster to move around without magic?

Patience, young "Padovan"

What is the command to reset a PC without deleting any files

DOS, create pipe for stdin/stdout of command.com(or 4dos.com) in C or Batch?

declaring a variable twice in IIFE

Why was the small council so happy for Tyrion to become the Master of Coin?

New order #4: World

What would happen to a modern skyscraper if it rains micro blackholes?

When blogging recipes, how can I support both readers who want the narrative/journey and ones who want the printer-friendly recipe?

Circuitry of TV splitters

A function which translates a sentence to title-case

A Journey Through Space and Time

Draw simple lines in Inkscape

Japan - Any leeway for max visa duration due to unforeseen circumstances?

least quadratic residue under GRH: an EXPLICIT bound

Download, install and reboot computer at night if needed

Is it legal to have the "// (c) 2019 John Smith" header in all files when there are hundreds of contributors?

Is it possible to do 50 km distance without any previous training?

Can a German sentence have two subjects?

A newer friend of my brother's gave him a load of baseball cards that are supposedly extremely valuable. Is this a scam?

Email Account under attack (really) - anything I can do?

Concept of linear mappings are confusing me

Turn Spark Stream from Socket into DataFrame

How to randomly select an item from a list?How to process a YAML stream in PythonWhy is reading lines from stdin much slower in C++ than Python?Delete column from pandas DataFrame by column nameHow to iterate over rows in a DataFrame in Pandas?Select rows from a DataFrame based on values in a column in pandasGet list from pandas DataFrame column headersCalculate rate of data processing from a Spark (Structured) Streaming ApplicationSpark Streaming - Identify Incoming File FormatAdding a column to a pandas dataframe based on cell values

.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty height:90px;width:728px;box-sizing:border-box;

0

I established a socket connection with my SparkSession that is sending a row of a .csv-File to my Stream.

So far my (PySpark-) code looks like this:

stream = spark.readStream.format('socket').option('host', 'localhost').option('port', 5555).load()

stream.writeStream.format('console').start().awaitTermination()

This prints the lines of the .csv File in one column like this:

+-----------------+
| value|
+-----------------+
|[2, C4653, C5030]|
+-----------------+

But what I actually would like to have is this:

+-----+-----+-----+
| col1| col2| col3|
+-----+-----+-----+
| 2|C4653|C5030|
+-----+-----+-----+

I would like to use this as a DataFrame to feed a ML-Pipeline with.

How can I process the incoming Stream Data?

python pyspark spark-streaming

share|improve this question

edited Mar 20 at 20:38

dnks23

asked Mar 20 at 19:34

dnks23dnks23

899

add a comment | 

0

I established a socket connection with my SparkSession that is sending a row of a .csv-File to my Stream.

So far my (PySpark-) code looks like this:

stream = spark.readStream.format('socket').option('host', 'localhost').option('port', 5555).load()

stream.writeStream.format('console').start().awaitTermination()

This prints the lines of the .csv File in one column like this:

+-----------------+
| value|
+-----------------+
|[2, C4653, C5030]|
+-----------------+

But what I actually would like to have is this:

+-----+-----+-----+
| col1| col2| col3|
+-----+-----+-----+
| 2|C4653|C5030|
+-----+-----+-----+

I would like to use this as a DataFrame to feed a ML-Pipeline with.

How can I process the incoming Stream Data?

python pyspark spark-streaming

share|improve this question

edited Mar 20 at 20:38

dnks23

asked Mar 20 at 19:34

dnks23dnks23

899

add a comment | 

I established a socket connection with my SparkSession that is sending a row of a .csv-File to my Stream.

So far my (PySpark-) code looks like this:

stream = spark.readStream.format('socket').option('host', 'localhost').option('port', 5555).load()

stream.writeStream.format('console').start().awaitTermination()

This prints the lines of the .csv File in one column like this:

+-----------------+
| value|
+-----------------+
|[2, C4653, C5030]|
+-----------------+

But what I actually would like to have is this:

+-----+-----+-----+
| col1| col2| col3|
+-----+-----+-----+
| 2|C4653|C5030|
+-----+-----+-----+

I would like to use this as a DataFrame to feed a ML-Pipeline with.

How can I process the incoming Stream Data?

python pyspark spark-streaming

share|improve this question

edited Mar 20 at 20:38

dnks23

asked Mar 20 at 19:34

dnks23dnks23

899

stream = spark.readStream.format('socket').option('host', 'localhost').option('port', 5555).load()

stream.writeStream.format('console').start().awaitTermination()

+-----------------+
| value|
+-----------------+
|[2, C4653, C5030]|
+-----------------+

+-----+-----+-----+
| col1| col2| col3|
+-----+-----+-----+
| 2|C4653|C5030|
+-----+-----+-----+

share|improve this question

edited Mar 20 at 20:38

dnks23

asked Mar 20 at 19:34

dnks23dnks23

899

share|improve this question

edited Mar 20 at 20:38

dnks23

asked Mar 20 at 19:34

dnks23dnks23

899

asked Mar 20 at 19:34

dnks23dnks23

899

asked Mar 20 at 19:34

dnks23dnks23

899

1 Answer
1

active

oldest

votes

1

You already has a Data Frame stream, which just need to change schema.

Just add this transformation after load() call:

 stream.selectExpr("split(value, ' ')[0] as col1","split(value, ' ')[1] as col2", "split(value, ' ')[2] as col3")

share|improve this answer

edited Mar 22 at 0:57

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  thanks for the answer, but trying .foreachRDD I get an error saying that DataFrame object has no attribute 'foreachRDD ? any hints on how I can get my incoming string into the desired format?
  
  – dnks23
  Mar 21 at 7:46
  

  
  
  
  


• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  I considered you are using stream from Spark Streaming. Since you already has Data Frame, it become easier. Updated the answer.
  
  – Volodymyr Zubariev
  Mar 22 at 0:58
  

  
  
  
  

add a comment | 

Your Answer

StackExchange.ifUsing("editor", function ()
StackExchange.using("externalEditor", function ()
StackExchange.using("snippets", function ()
StackExchange.snippets.init();
);
);
, "code-snippets");

StackExchange.ready(function()
var channelOptions =
tags: "".split(" "),
id: "1"
;
initTagRenderer("".split(" "), "".split(" "), channelOptions);

StackExchange.using("externalEditor", function()
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled)
StackExchange.using("snippets", function()
createEditor();
);

else
createEditor();

);

function createEditor()
StackExchange.prepareEditor(
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader:
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
,
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
);



);

draft saved

draft discarded

Sign up or log in

StackExchange.ready(function ()
StackExchange.helpers.onClickDraftSave('#login-link');
);

Sign up using Google

Sign up using Facebook

Sign up using Email and Password

Post as a guest

Name

Email

Required, but never shown

StackExchange.ready(
function ()
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f55268874%2fturn-spark-stream-from-socket-into-dataframe%23new-answer', 'question_page');

);

Post as a guest

Name

Email

Required, but never shown

1

You already has a Data Frame stream, which just need to change schema.

Just add this transformation after load() call:

 stream.selectExpr("split(value, ' ')[0] as col1","split(value, ' ')[1] as col2", "split(value, ' ')[2] as col3")

share|improve this answer

edited Mar 22 at 0:57

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  thanks for the answer, but trying .foreachRDD I get an error saying that DataFrame object has no attribute 'foreachRDD ? any hints on how I can get my incoming string into the desired format?
  
  – dnks23
  Mar 21 at 7:46
  

  
  
  
  


• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  I considered you are using stream from Spark Streaming. Since you already has Data Frame, it become easier. Updated the answer.
  
  – Volodymyr Zubariev
  Mar 22 at 0:58
  

  
  
  
  

add a comment | 

1

You already has a Data Frame stream, which just need to change schema.

Just add this transformation after load() call:

 stream.selectExpr("split(value, ' ')[0] as col1","split(value, ' ')[1] as col2", "split(value, ' ')[2] as col3")

share|improve this answer

edited Mar 22 at 0:57

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  thanks for the answer, but trying .foreachRDD I get an error saying that DataFrame object has no attribute 'foreachRDD ? any hints on how I can get my incoming string into the desired format?
  
  – dnks23
  Mar 21 at 7:46
  

  
  
  
  


• 
  
  
  

  
  

  
  
  
  
  
  

  
  
  I considered you are using stream from Spark Streaming. Since you already has Data Frame, it become easier. Updated the answer.
  
  – Volodymyr Zubariev
  Mar 22 at 0:58
  

  
  
  
  

add a comment | 

You already has a Data Frame stream, which just need to change schema.

Just add this transformation after load() call:

 stream.selectExpr("split(value, ' ')[0] as col1","split(value, ' ')[1] as col2", "split(value, ' ')[2] as col3")

share|improve this answer

edited Mar 22 at 0:57

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

 stream.selectExpr("split(value, ' ')[0] as col1","split(value, ' ')[1] as col2", "split(value, ' ')[2] as col3")

share|improve this answer

edited Mar 22 at 0:57

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188

answered Mar 20 at 20:48

Volodymyr ZubarievVolodymyr Zubariev

1188