<jupyter_start><jupyter_text>Load small dataset<jupyter_code>dataset = spark.read.csv('dataset.csv')
feature = open("feature.name","r", encoding='utf-8').read()
feature = re.sub('[!@#$]', '', feature)
col_name = feature.split("\t")
col_name.append("retweets")
dataset = dataset.toDF(*col_name)
dataset.show(3)
# from statistics import pvariance
# from statistics import median
# from statistics import mean
def cleaning(data):
    now = datetime.now()
    
    data = data.withColumn("days", date_format(unix_timestamp(data["timestamp"],  "EEE MMM dd HH:mm:ss Z yyyy") \
                                                .cast(TimestampType()),"EEE")) # weekdays
    
    stringIndexer = StringIndexer().setInputCol("days").setOutputCol("weeks") # weekdays to stringIndex
    data = stringIndexer.fit(data).transform(data)
    
    
    data = data.withColumn("hours", date_format(unix_timestamp(data["timestamp"],  "EEE MMM dd HH:mm:ss Z yyyy") \
                                                .cast(TimestampType()),"HH").cast("int")) # hours of day
    
    def q_hours(x):
        if x <= 5: 
            return 1.0  #0 -> 5
        elif x > 5 and x <=11: 
            return 2.0 #6 -> 11
        elif x > 11 and x <=17:
            return 3.0  # 12 -> 17
        else:
            return 4.0 # 18-> 23
    q_hour = udf(q_hours, FloatType())
    data = data.withColumn("q_hours", q_hour(col("hours"))) # 4 type of hours in a day
    
    
    def to_log(x):
        if x == None:
            return 0.0
        else:
            return x

    log_to = udf(to_log, DoubleType())
    
    data = data.withColumn('followers', log_to(log10(col('followers'))))
    data = data.withColumn('friends', log_to(log10(col('friends'))))
    data = data.withColumn('favorites', log_to(log10(col('favorites'))))
    data = data.withColumn('retweets', log_to(log10(col('retweets'))))
    
    data = data.withColumn("posi_sen", split(data['sentiment'], ' ').getItem(0).cast('float'))
    data = data.withColumn("nega_sen", split(data['sentiment'], ' ').getItem(1).cast('float'))
   

    def ent_count(row):
        lis = []
        for x in row:
            try:
                if x[0].isupper() == True:
                    #x = re.sub('[!@#$%]', '', x)
                    lis.append(x)
            except:
                continue
        if len(lis) == 0:
            return 0.0
        return float(len(lis)) #sum(lis) / len(lis)

    count_ent = udf(ent_count, FloatType())
    data = data.withColumn("count_ent_an", count_ent(split('entities', '[;:]')))

    def enti_score(row):
        lis = []
        for x in row:
            try:
                lis.append(float(x))
            except:
                continue
        if len(lis) == 0:
            return 0.0
        return sum(lis) / len(lis)
    
    cal = udf(enti_score, FloatType())
    data = data.withColumn("ent_score", cal(split('entities', '[;:]')))

    
#     hasher = FeatureHasher(inputCols=["hashtags",'mentions'],
#                        outputCol="hasher")
#     data = hasher.transform(data)


    
    data = data.select("followers","friends","favorites","posi_sen","nega_sen",
                       "count_ent_an","ent_score","days","weeks","hours","q_hours", 
                       "retweets")
        
    print(data.printSchema())
    print(datetime.now()-now)
    return data
dataset = cleaning(dataset)
#dataset.select("hours","hours1").show()
dataset.filter(dataset["hours"]== 17).select("q_hours").show(2)
dataset.filter(dataset['days']!="Tue").select("days").show(2)
# df.filter(df['weeks']!=3) \
# .filter(df['weeks']!=0) \
# .filter(df['weeks']!=2) \
# .filter(df['weeks']!=1) \
# .filter(df['weeks']!=6) \
# .filter(df['weeks']!=5) \
# .filter(df['weeks']!=4) \
# .select("weeks").show(10)<jupyter_output><empty_output><jupyter_text>Linear Regression<jupyter_code>three = VectorAssembler().setInputCols(['followers',"friends","favorites","posi_sen",
                                            "nega_sen","count_ent_an","ent_score","weeks","q_hours"])\
        .setOutputCol("features")
data = three.transform(dataset)
data.printSchema()
train, test = data.randomSplit(weights=[0.8,0.2], seed=42)
from pyspark.ml import Pipeline
from pyspark.ml.regression import LinearRegression
### without pipeline
lr = LinearRegression(featuresCol="features", labelCol='retweets')
now = datetime.now()
lrmodel = lr.fit(train)
print(datetime.now()-now)
pred = lrmodel.transform(test)
pred.select("features","retweets","prediction").show(10)
from pyspark.ml.evaluation import RegressionEvaluator
regressionEvaluator = RegressionEvaluator(predictionCol="prediction", labelCol="retweets",metricName='rmse')
rmse = regressionEvaluator.evaluate(pred)
print(f"RMSE is {rmse:.1f}")
r2 = regressionEvaluator.setMetricName('r2').evaluate(pred)
print(f"R2 is {r2}")<jupyter_output>R2 is 0.9087026295443565<jupyter_text>Random Forest<jupyter_code>train, test = data.randomSplit(weights=[0.8,0.2], seed=42)
from pyspark.ml import Pipeline
from pyspark.ml.regression import RandomForestRegressor
rf = RandomForestRegressor(labelCol="retweets", maxBins=100, seed=42)
rfmodel = rf.fit(train)
pred = rfmodel.transform(test)
regressionEvaluator = RegressionEvaluator(predictionCol="prediction", labelCol="retweets",metricName='rmse')
rmse = regressionEvaluator.evaluate(pred)
print(f"RMSE is {rmse:.1f}")
r2 = regressionEvaluator.setMetricName('r2').evaluate(pred)
print(f"R2 is {r2}")<jupyter_output>R2 is 0.8543836339798685<jupyter_text>Cross validation<jupyter_code>from pyspark.ml.tuning import ParamGridBuilder
paramGrid = (ParamGridBuilder()
    .addGrid(rf.maxDepth, [2, 4, 6])
    .addGrid(rf.numTrees, [10, 100])
    .build())
evaluator = RegressionEvaluator(labelCol="retweets",
    predictionCol="prediction",
    metricName="r2")
from pyspark.ml.tuning import CrossValidator
cv = CrossValidator(estimator=rf,
 evaluator=evaluator,
 estimatorParamMaps=paramGrid,
 numFolds=3,
 seed=42)
cvModel = cv.fit(test)
list(zip(cvModel.getEstimatorParamMaps(), cvModel.avgMetrics))<jupyter_output><empty_output><jupyter_text>Tensorflow<jupyter_code>import pandas as pd
now = datetime.now()
df = dataset.toPandas()
df.to_csv('tensor.csv', index=False)
print(datetime.now() - now)
df = pd.read_csv('tensor.csv')
df.head()
del df['days']
del df['hours']
df.info()
x = df.iloc[:, 0:-1].values.astype(float)
y = df.iloc[:, -1].values.astype(float)
from sklearn.preprocessing import StandardScaler
sc = StandardScaler()
x = sc.fit_transform(x)
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(x, y, test_size = 0.2, random_state = 42)
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, BatchNormalization
import tensorflow.keras.backend as K

# https://stackoverflow.com/questions/42665359/how-do-you-compute-accuracy-in-a-regression-model-after-rounding-predictions-to
def soft_acc(y_true, y_pred):
    return K.mean(K.equal(K.round(y_true), K.round(y_pred)))
def Model():
    
    model = Sequential([
        Dense(90, input_shape=(9,), activation="relu"),
        BatchNormalization(),
        Dense(162, activation='relu'),
        BatchNormalization(),
        Dense(162, activation='relu'),
        BatchNormalization(),
        Dense(63, activation='relu'),
        BatchNormalization(),
        Dense(32, activation='relu'),
        BatchNormalization(),
        Dense(1)
        
    ])
    
#     model.compile(optimizer = 'Nadam', loss = 'mse', metrics = ['accuracy'])
    model.compile(optimizer = 'adam', loss = 'mse', metrics = [soft_acc])
    return model
model = Model()
history = model.fit(X_train, y_train, validation_split=0.2, shuffle=True,
                         validation_data=(X_test,y_test), 
                         epochs=100, batch_size=1000)
%matplotlib inline
import matplotlib.pyplot as plt
#import seaborn as sns

def plots(acc, val_acc, loss, val_loss):
    fig, (ax1, ax2) = plt.subplots(ncols=2, sharey=False)

    # plot accuracy
    ax1.plot(acc, label='Training')
    if val_acc is not None:
        ax1.plot(val_acc, label='Validation')

    ax1.set_xlabel('Epoch')
    ax1.set_title('Model Accuracy')
    ax1.legend(loc='upper left')
    ax1.set_ylim([0,1])

    # plot loss
    ax2.plot(loss, label='Training')
    if val_loss is not None:
        ax2.plot(val_loss, label='Validation')

    ax2.set_xlabel('Epoch')
    ax2.set_title('Model Loss')
    ax2.legend(loc='upper right')
    ax2.set_ylim([0,1])

    plt.show()

plt.rcParams["figure.figsize"] = (13,4)
plots(history.history["soft_acc"],history.history['val_soft_acc'],
      history.history["loss"],history.history['val_loss'])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15922396969349098, 'fraction_numerical': 0.023189071289174606, 'mean_word_length': 3.5186721991701244, 'pattern_counts': {'":': 0, '<': 22, '<?xml version=': 0, '>': 25, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '147863', 'license': 'MIT', 'n_tokens_mistral': 3139, 'n_tokens_neox': 2956, 'n_words': 516, 'path': 'retweet/retweet_mock_up.ipynb', 'repo_name': 'tardatio/retweet', 'size': '8711'}