如何使用Python从MongoDB创建LabeledPoint？附CSV实现代码

Hey there! Let's adapt your existing LabeledPoint code to work with the DataFrame you've pulled from MongoDB. Here's a step-by-step breakdown:

1. Complete the MongoDB to DataFrame Import #

First, let's finish that MongoDB connection code to properly load your collection into a Pandas DataFrame:

from pymongo import MongoClient
import pandas as pd

# Connect to MongoDB (fill in your full connection string if needed)
client = MongoClient('mongodb://localhost:27017/')
db = client['your_database_name']  # Replace with your actual database name
collection = db['your_collection_name']  # Replace with your actual collection name

# Convert MongoDB collection to a Pandas DataFrame
df = pd.DataFrame(list(collection.find()))

2. Update the LabeledPoint Creation Function #

Your original function works with a flat list (from CSV rows), but we need to adjust it to handle Pandas Series (rows from the DataFrame). Also, note a small quirk in your original code: you reference q3 but never define it—assuming that's a typo (maybe it should pull from fields[2]), I'll adjust for that here. You can use either column names (for readability) or positional indexing (like your CSV code):

from pyspark.mllib.regression import LabeledPoint
import numpy as np

def createLabeledPoints(row):
    # Option 1: Use column names (swap with your actual column names)
    q1 = int(row['q1_column'])
    q2 = int(row['q2_column'])
    q3 = int(row['q3_column'])  # Fixed the missing q3 definition from your original code
    result = int(row['result_column'])

    # Option 2: Use positional indexing (matching your CSV approach)
    # q1 = int(row.iloc[0])
    # q2 = int(row.iloc[1])
    # q3 = int(row.iloc[2])
    # result = int(row.iloc[38])

    return LabeledPoint(result, np.array([q1, q2, q3]))

3. Generate LabeledPoints from the DataFrame #

Now apply the function to every row in your DataFrame to create a list of LabeledPoints:

# Apply the function across all rows (axis=1 tells Pandas to process row-by-row)
labeled_points = df.apply(createLabeledPoints, axis=1).tolist()

That's it! You'll now have a list of LabeledPoint objects ready for your Spark ML workflows, just like you had with the CSV data.

内容的提问来源于stack exchange，提问作者salma