Pyspark - How to Find the consecutive values in PySpark DataFrame column and replace the value

#Pyspark - How to Find the consecutive values in PySpark DataFrame column and replace the value

from pyspark.sql.functions import *
from pyspark.sql.window import Window
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()
data1 = [
    (1, 'ABC'),
    (2, 'ABC'),
    (3, 'ABC'),
    (4, 'XYZ'),
    (5, 'PQR'),
    (6, 'PQR')
]
columns1 = ['id', 'dept']

dataDF = spark.createDataFrame(data=data1, schema=columns1)
dataDF.show()
winSpec = Window.partitionBy('dept').orderBy('id')

dataDF.withColumn('row_num', row_number().over(winSpec))\
    .withColumn('result', col("dept"))\
    .show(truncate=False)

# +---+----+-------+------+
# |id |dept|row_num|result|
# +---+----+-------+------+
# |5  |PQR |1      |PQR   |
# |6  |PQR |2      |PQR   |
# |4  |XYZ |1      |XYZ   |
# |1  |ABC |1      |ABC   |
# |2  |ABC |2      |ABC   |
# |3  |ABC |3      |ABC   |
# +---+----+-------+------+


dataDF.withColumn('row_num', row_number().over(winSpec))\
    .withColumn('result', when((col("row_num") > 1), concat_ws('_', "dept", "row_num")).otherwise(col("dept")))\
    .show(truncate=False)

# +---+----+-------+------+
# |id |dept|row_num|result|
# +---+----+-------+------+
# |5  |PQR |1      |PQR   |
# |6  |PQR |2      |PQR_2 |
# |4  |XYZ |1      |XYZ   |
# |1  |ABC |1      |ABC   |
# |2  |ABC |2      |ABC_2 |
# |3  |ABC |3      |ABC_3 |
# +---+----+-------+------+

#But we supposed to get output as below
# We only have to add _1/_2 when duplicate is there
# use lag/lead to find out duplicate present
# +---+----+-------+------+
# |id |dept|row_num|result|
# +---+----+-------+------+
# |5  |PQR |1      |PQR_1 |
# |6  |PQR |2      |PQR_2 |
# |4  |XYZ |1      |XYZ   |
# |1  |ABC |1      |ABC_1 |
# |2  |ABC |2      |ABC_2 |
# |3  |ABC |3      |ABC_3 |
# +---+----+-------+------+

wind_id = Window.orderBy('id')
wind_by_dept = Window.partitionBy("dept").orderBy("dept")
condition = col("chk_duplicate")| ((col("chk_length")>1) & (col("row_num")==1))

new_df = dataDF.withColumn("chk_duplicate",col("dept")==lag("dept").over(wind_id))\
.withColumn("row_num",row_number().over(wind_by_dept))\
.withColumn("chk_length",count("dept").over(wind_by_dept))\
.withColumn("dept_updated",when(condition,concat_ws("_", *["dept","row_num"]))
                                                   .otherwise(col("dept")))

new_df.orderBy("id").show()
# +---+----+-------------+-------+----------+------------+
# | id|dept|chk_duplicate|row_num|chk_length|dept_updated|
# +---+----+-------------+-------+----------+------------+
# |  1| ABC|         null|      1|         3|       ABC_1|
# |  2| ABC|         true|      2|         3|       ABC_2|
# |  3| ABC|         true|      3|         3|       ABC_3|
# |  4| XYZ|        false|      1|         1|         XYZ|
# |  5| PQR|        false|      1|         2|       PQR_1|
# |  6| PQR|         true|      2|         2|       PQR_2|
# +---+----+-------------+-------+----------+------------+

#*dataDF.columns, dept_updated - having issue - keep wild card later
#only named arguments may follow *expression
#new_df.select(*dataDF.columns, 'dept_updated').orderBy("id").show()
new_df.select('dept_updated', *dataDF.columns).orderBy("id").show()

# +------------+---+----+
# |dept_updated| id|dept|
# +------------+---+----+
# |       ABC_1|  1| ABC|
# |       ABC_2|  2| ABC|
# |       ABC_3|  3| ABC|
# |         XYZ|  4| XYZ|
# |       PQR_1|  5| PQR|
# |       PQR_2|  6| PQR|
# +------------+---+----+

#or
new_df.select('id', 'dept', 'dept_updated').orderBy("id").show()
# +---+----+------------+
# | id|dept|dept_updated|
# +---+----+------------+
# |  1| ABC|       ABC_1|
# |  2| ABC|       ABC_2|
# |  3| ABC|       ABC_3|
# |  4| XYZ|         XYZ|
# |  5| PQR|       PQR_1|
# |  6| PQR|       PQR_2|
# +---+----+------------+

Pyspark - concat list of all items per day

 #Pyspark - concat list of all items per day

from pyspark.sql.functions import *
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()

data1 = [
    ('2021-02-22', 'cricket_bat'),
    ('2021-02-22', 'cricket_ball'),
    ('2021-02-22', 'cricket_glove'),
    ('2021-02-23', 'shuttle_cock'),
    ('2021-02-24', 'shuttle_racket')
]

columns1 = ['date', 'product']
dataDF = spark.createDataFrame(data=data1, schema=columns1)
dataDF.show()

dataDF.groupBy('date').agg(count('product').alias('count'),
                           collect_list('product').alias('list'),
                           concat_ws(',', collect_list('product')).alias('concat')
                           ).show(truncate=False)

Output:

Pyspark - Find sum of login time for each employee per each day

 #Pyspark - Find sum of login time for each employee per each day

from pyspark.sql.functions import *
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()

data1 = [(1, '2021-02-22', 5, 20),
(1, '2021-02-22', 50, 445),
(1, '2021-02-22', 500, 575),
(2, '2021-02-23', 15, 70),
(3, '2021-02-24', 45, 95),
(4, '2021-02-24', 100, 300)]

columns1 = ['emp_id', 'date', 'in', 'out']
dataDF = spark.createDataFrame(data=data1, schema=columns1)
dataDF.show()

dataDF.groupBy('emp_id', 'date')\
    .agg(sum(col('out') - col('in'))).alias('diff').orderBy('emp_id', 'date').show()

Output:

Pyspark - Last login for each employee & date

 #Last login for each employee & date

from pyspark.sql.functions import *
from pyspark.sql.window import Window
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()

data1 = [(1, '2021-02-22 00:00:00'),
(1, '2021-02-22 09:00:00'),
(1, '2021-02-22 11:00:00'),
(2, '2021-02-23 11:00:00'),
(2, '2021-02-24 23:00:00'),
(2, '2021-02-24 23:15:00')]

columns1 = ['emp_id', 'date_time_col']
dataDF = spark.createDataFrame(data=data1, schema=columns1)
dataDF.show()

dataDF.groupBy('emp_id', to_date('date_time_col', 'yyyy-MM-dd').alias('calendar_day'))\
    .agg(max('date_time_col').alias('last_login_time_by_day')).show()


Output:

Find Employee whose salary greater than the Manager Salary

 #EMP Salary > Manager Salary

from pyspark.sql.functions import *
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()
data1 = [
(1, "John", 35000, None),
(2, "Peter", 45000, 1),
(3, "Sam", 5000, 2),
(4, "Ramu", 55000, 2)]
columns1 = ['id', 'name', 'salary', 'mgr_id']
empDF = spark.createDataFrame(data = data1, schema = columns1)
empDF.show()

empDF.alias('E').join(empDF.alias('M'), on=[col("E.mgr_id") == col("M.id")], how='inner')\
    .filter("E.salary > M.salary").show()

Output:

Pyspark Joins

data1_cols = ["id","name"]

data1 = [(1, "Sugreeva"), (2, "Ravan"), (4, "Hanuman"), (4, "Hanuman"), (2, "Ravan")]

data1_df = spark.createDataFrame(data = data1, schema = data1_cols)

data1_df.show(truncate=False)

+---+--------+
| id|    name|
+---+--------+
|  1|Sugreeva|
|  2|   Ravan|
|  4| Hanuman|
|  4| Hanuman|
|  2|   Ravan|
+---+--------+

data2_cols = ["id","name"]

data2 = [(2, "Ravan"), (2, "Ravan"), (1, "Sugreeva"), (6, "Ram"), (7, "Sita"), (7, "Sita")]

data2_df = spark.createDataFrame(data = data2, schema = data2_cols)

data2_df.show(truncate=False)

+---+--------+
| id|    name|
+---+--------+
|  2|   Ravan|
|  2|   Ravan|
|  1|Sugreeva|
|  6|     Ram|
|  7|    Sita|
|  7|    Sita|
+---+--------+

#inner (M*N if any duplicates)

data1_df.join(data2_df, on=[data1_df.id == data2_df.id], how="inner").show(truncate=False)

+---+--------+---+--------+
|id |name    |id |name    |
+---+--------+---+--------+
|1  |Sugreeva|1  |Sugreeva|
|2  |Ravan   |2  |Ravan   |
|2  |Ravan   |2  |Ravan   |
|2  |Ravan   |2  |Ravan   |
|2  |Ravan   |2  |Ravan   |
+---+--------+---+--------+

#left join (M*N if any duplicates)

data1_df.join(data2_df, on=[data1_df.id == data2_df.id], how="left").show(truncate=False)

+---+--------+----+--------+
|id |name    |id  |name    |
+---+--------+----+--------+
|1  |Sugreeva|1   |Sugreeva|
|2  |Ravan   |2   |Ravan   |
|2  |Ravan   |2   |Ravan   |
|2  |Ravan   |2   |Ravan   |
|2  |Ravan   |2   |Ravan   |
|4  |Hanuman |null|null    |
|4  |Hanuman |null|null    |
+---+--------+----+--------+

#right join (M*N if any duplicates)

data1_df.join(data2_df, on=[data1_df.id == data2_df.id], how="right").show(truncate=False)

+----+--------+---+--------+
|id  |name    |id |name    |
+----+--------+---+--------+
|null|null    |7  |Sita    |
|null|null    |7  |Sita    |
|null|null    |6  |Ram     |
|1   |Sugreeva|1  |Sugreeva|
|2   |Ravan   |2  |Ravan   |
|2   |Ravan   |2  |Ravan   |
|2   |Ravan   |2  |Ravan   |
|2   |Ravan   |2  |Ravan   |
+----+--------+---+--------+

#left_anti

data1_df.join(data2_df, on=[data1_df.id == data2_df.id], how="left_anti").show(truncate=False)

+---+-------+
|id |name   |
+---+-------+
|4  |Hanuman|
|4  |Hanuman|
+---+-------+

#left_semi

data1_df.join(data2_df, on=[data1_df.id == data2_df.id], how="left_semi").show(truncate=False)

+---+--------+
|id |name    |
+---+--------+
|1  |Sugreeva|
|2  |Ravan   |
|2  |Ravan   |
+---+--------+

#right_anti (not by default, reverse join dataframes & use left_anti)

data2_df.join(data1_df, on=[data1_df.id == data2_df.id], how="left_anti").show(truncate=False)

+---+----+
|id |name|
+---+----+
|7  |Sita|
|7  |Sita|
|6  |Ram |
+---+----+

pyspark join condition (2 types)

 columns1 = ["emp_id","emp_name","emp_city","emp_salary"]

data1 = [
(1, "John", "Sydney", 35000.00),
(2, "Peter", "Melbourne", 45000.00),
(3, "Sam", "Sydney", 55000.00)]
emp_df = spark.createDataFrame(data = data1, schema = columns1)
emp_df.show(truncate=False)
emp_df.printSchema()

data2 = [
(2, "Peter", "Melbourne", 55000.00),
(5, "Jessie", "Brisbane", 42000.00)]
emp_delta_df = spark.createDataFrame(data = data2, schema = columns1)
emp_delta_df.show(truncate=False)
emp_delta_df.printSchema()

# emp_id from emp_delta_df not shown
print('### emp_id from emp_delta_df not shown')
emp_df.join(emp_delta_df, "emp_id", "inner").show()

# emp_id is shown from both dataframes
print('### emp_id is repeated from both dataframes')
emp_df.join(emp_delta_df, emp_df.emp_id == emp_delta_df.emp_id, "inner").show()

### emp_id from emp_delta_df not shown
+------+--------+---------+----------+--------+---------+----------+
|emp_id|emp_name| emp_city|emp_salary|emp_name| emp_city|emp_salary|
+------+--------+---------+----------+--------+---------+----------+
|     2|   Peter|Melbourne|   45000.0|   Peter|Melbourne|   55000.0|
+------+--------+---------+----------+--------+---------+----------+

### emp_id is repeated from both dataframes
+------+--------+---------+----------+------+--------+---------+----------+
|emp_id|emp_name| emp_city|emp_salary|emp_id|emp_name| emp_city|emp_salary|
+------+--------+---------+----------+------+--------+---------+----------+
|     2|   Peter|Melbourne|   45000.0|     2|   Peter|Melbourne|   55000.0|
+------+--------+---------+----------+------+--------+---------+----------+

pyspark scd type2 with delta file

 from pyspark.sql.functions import *

from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local')\
                .appName('scdType2')\
                .getOrCreate()
emp_df = spark.read.csv('C:\\Users\\Prabhath\\Downloads\\employee.csv', inferSchema=True, header=True,
                        ignoreLeadingWhiteSpace=True, ignoreTrailingWhiteSpace=True)
emp_df = emp_df.alias('emp_df')
emp_df.show()

emp_delta_df = spark.read.csv('C:\\Users\\Prabhath\\Downloads\\employee_delta.csv', inferSchema=True, header=True,
                        ignoreLeadingWhiteSpace=True, ignoreTrailingWhiteSpace=True)
emp_delta_df = emp_delta_df.alias('emp_delta_df')
emp_delta_df.show()

#in-active
emp_inactive = emp_df.join(emp_delta_df, emp_df.emp_id==emp_delta_df.emp_id, "inner")\
    .select("emp_df.*")\
    .withColumn('is_active', lit(False))\
    .withColumn('start_time', current_timestamp())\
    .withColumn('end_time', current_timestamp())
print('### inactive records - common in both - emp_df depreated, so take only those')
emp_inactive.show(truncate=False)

#active
emp_active = emp_df.join(emp_delta_df, emp_df.emp_id==emp_delta_df.emp_id, "inner")\
    .select("emp_delta_df.*")\
    .withColumn('is_active', lit(True))\
    .withColumn('start_time', current_timestamp())\
    .withColumn('end_time', to_timestamp(lit('9999-12-31 00:00:00.000'), 'yyyy-MM-dd HH:mm:ss.SSSS'))
print('### active records - common in both - emp_delta_df latest, so take only those')
emp_active.show(truncate=False)

#only left (no
emp_no_change_df = emp_df.join(emp_delta_df, emp_df.emp_id==emp_delta_df.emp_id, "left_anti")\
    .select("emp_df.*")\
    .withColumn('is_active', lit(True))\
    .withColumn('start_time', current_timestamp())\
    .withColumn('end_time', to_timestamp(lit('9999-12-31 00:00:00.000'), 'yyyy-MM-dd HH:mm:ss.SSSS'))
print('### not changed records - left anti - emp_df join emp_delta_df')
emp_no_change_df.show(truncate=False)


#only right (new)
emp_new_df = emp_delta_df.join(emp_df, emp_df.emp_id==emp_delta_df.emp_id, "left_anti")\
    .select("emp_delta_df.*")\
    .withColumn('is_active', lit(True))\
    .withColumn('start_time', current_timestamp())\
    .withColumn('end_time', to_timestamp(lit('9999-12-31 00:00:00.000'), 'yyyy-MM-dd HH:mm:ss.SSSS'))
print('### new records - left anti - emp_delta_df join emp_df')
emp_new_df.show(truncate=False)

#union all
final_df = emp_inactive.union(emp_active).union(emp_no_change_df).union(emp_new_df).orderBy('emp_id', 'is_active')
final_df.show(truncate=False)

+------+--------+---------+----------+
|emp_id|emp_name| emp_city|emp_salary|
+------+--------+---------+----------+
|     1|    John|   Sydney|     35000|
|     2|   Peter|Melbourne|     45000|
|     3|     Sam|   Sydeny|     55000|
+------+--------+---------+----------+

+------+--------+---------+----------+
|emp_id|emp_name| emp_city|emp_salary|
+------+--------+---------+----------+
|     2|   Peter|Melbourne|     55000|
|     4|     Sam|   Sydeny|     42000|
+------+--------+---------+----------+

### inactive records - common in both - emp_df depreated, so take only those
+------+--------+---------+----------+---------+-----------------------+-----------------------+
|emp_id|emp_name|emp_city |emp_salary|is_active|start_time             |end_time               |
+------+--------+---------+----------+---------+-----------------------+-----------------------+
|2     |Peter   |Melbourne|45000     |false    |2023-04-05 21:28:26.006|2023-04-05 21:28:26.006|
+------+--------+---------+----------+---------+-----------------------+-----------------------+

### active records - common in both - emp_delta_df latest, so take only those
+------+--------+---------+----------+---------+----------------------+-------------------+
|emp_id|emp_name|emp_city |emp_salary|is_active|start_time            |end_time           |
+------+--------+---------+----------+---------+----------------------+-------------------+
|2     |Peter   |Melbourne|55000     |true     |2023-04-05 21:28:26.36|9999-12-31 00:00:00|
+------+--------+---------+----------+---------+----------------------+-------------------+

### not changed records - left anti - emp_df join emp_delta_df
+------+--------+--------+----------+---------+-----------------------+-------------------+
|emp_id|emp_name|emp_city|emp_salary|is_active|start_time             |end_time           |
+------+--------+--------+----------+---------+-----------------------+-------------------+
|1     |John    |Sydney  |35000     |true     |2023-04-05 21:28:26.586|9999-12-31 00:00:00|
|3     |Sam     |Sydeny  |55000     |true     |2023-04-05 21:28:26.586|9999-12-31 00:00:00|
+------+--------+--------+----------+---------+-----------------------+-------------------+

### new records - left anti - emp_delta_df join emp_df
+------+--------+--------+----------+---------+-----------------------+-------------------+
|emp_id|emp_name|emp_city|emp_salary|is_active|start_time             |end_time           |
+------+--------+--------+----------+---------+-----------------------+-------------------+
|4     |Sam     |Sydeny  |42000     |true     |2023-04-05 21:28:26.781|9999-12-31 00:00:00|
+------+--------+--------+----------+---------+-----------------------+-------------------+

### union all (inactive, active, not_changed, changed)
+------+--------+---------+----------+---------+-----------------------+-----------------------+
|emp_id|emp_name|emp_city |emp_salary|is_active|start_time             |end_time               |
+------+--------+---------+----------+---------+-----------------------+-----------------------+
|1     |John    |Sydney   |35000     |true     |2023-04-05 21:28:26.956|9999-12-31 00:00:00    |
|2     |Peter   |Melbourne|45000     |false    |2023-04-05 21:28:26.956|2023-04-05 21:28:26.956|
|2     |Peter   |Melbourne|55000     |true     |2023-04-05 21:28:26.956|9999-12-31 00:00:00    |
|3     |Sam     |Sydeny   |55000     |true     |2023-04-05 21:28:26.956|9999-12-31 00:00:00    |
|4     |Sam     |Sydeny   |42000     |true     |2023-04-05 21:28:26.956|9999-12-31 00:00:00    |
+------+--------+---------+----------+---------+-----------------------+-----------------------+

Python Palindrome

Using Recursion

def isPalindrome(testVariable):
  print(testVariable)
  if len(testVariable) <= 1:
    return True
  
  length = len(testVariable)
  if testVariable[0] == testVariable[length-1]:
      return isPalindrome(testVariable[1:length-1])

  return False

a = isPalindrome('MADAM')
print(a)  ## True

Using normal way

def isPalindrome(testVariable):
  if testVariable == testVariable[::-1]:
    return True
  return False

a = isPalindrome('MADAM')
print(a)  ## True

Python Find second smallest number

Python Find second smallest number

def get_second_smallest(numbers):
    num1, num2 = float('inf'), float('inf')
    print num1, num2
    print '#####'
    for x in numbers:
        if x <= num1:
            num1, num2 = x, num1
            print num1, '-------', num2
        elif x < num2:
            num2 = x
            print num1, '-------', num2
        else:
            print 'pass... ' + str(x)
    return num2

out = get_second_smallest([2, 3, 4, 5, 1, 1.3, 1.5])
print '\nOutput: ' + str(out)

Output:
2 ------- inf
2 ------- 3
pass... 4
pass... 5
1 ------- 2
1 ------- 1.3
pass... 1.5

Output: 1.3

Replace string in all files in a directory in perl

We will discuss briefly about the script (To replace string in all files in a directory in perl)

Please define your directory path in the following :
my $input_dir  = "/usr/src/perl_test/input/";
my $output_dir = "/usr/src/perl_test/output/";
my $log_dir    = "/usr/src/perl_test/log/";

Define the string to replace and string to replace with as mentioned below
my $str_to_replace = "Mother Teresa":
my $str_replace_with = "MOTHER TERESA";

If you want to replace a particular string in all the files in the same directory path, then give the same path for both $input_dir & $output_dir.
E.g.,
my $input_dir  = "/usr/src/perl_test/same_path/";
my $output_dir = "/usr/src/perl_test/same_path/";

Log Directory :
It creates the log file(log_<timestamp>.log) in the path mentioned in $log_dir

Important Note : 
Please take a backup before running the script for safer side in case if you are replacing the strings in the same folder


conversion_script.pl

#!/usr/intel/bin/perl

use strict;
use warnings;
use Data::Dumper;

my $input_dir  = "/usr/src/perl_test/input/";
my $output_dir = "/usr/src/perl_test/output/";
my $log_dir    = "/usr/src/perl_test/log/";

my $str_to_replace = "Mother Teresa":
my $str_replace_with = "MOTHER TERESA";

my ($sec, $min, $hour, $mday, $mon, $year, $wday, $yday, $isdst);

($sec,$min,$hour,$mday,$mon,$year,$wday,$yday,$isdst) = localtime(time);
 $sec  = sprintf("%02d",$sec);
 $min  = sprintf("%02d",$min);
 $hour = sprintf("%02d",$hour);
 $mday = sprintf("%02d", $mday);
 $mon  = sprintf("%02d", $mon+1);
 $year = sprintf("%04d", $year+1900);

my $output_tag = $mday .'_'. $mon .'_'. $year .'_'. $hour .'_'. $min;

opendir(IN_DIR, $input_dir)   or die $!;
opendir(OUT_DIR, $output_dir) or die $!;
opendir(LOG_DIR, $log_dir)    or die $!;

open (LOG_FILE, "> $log_dirlog_$output_tag.log");

print LOG_FILE "------------- START ------------ \n";

#print LOG_FILE "Started Copying Files from Source: $srcdir   to Destination: $dest \n";
#my $cmd = "cp -R $srcdir/* $dest/";
#`$cmd`; #or die "Chk the command : " . $cmd;
#print LOG_FILE "Finished Copying Files from Source: $srcdir   to Destination: $dest \n";

my (%files_changed, %files_not_changed, @invalid_list_files);

while (my $file = readdir(IN_DIR)) {

    # Ignore if it is not file 
    unless (-f "$input_dir$file") {
       print LOG_FILE "-W- Not a File, Ignoring : " . $file . "\n";
       next;
    }

    print LOG_FILE "\n\nFile Name : " . $file . "\n";

    open(IN_FILE, "<$input_dir$file") || warn "Cant open file for reading: " . "$input_dir$file";
    my @lines = ;
    close(IN_FILE);

    my @newlines;
    foreach my $each_line (@lines) {
        #chomp $each_line;
        if($each_line =~ /$str_to_replace/) {
            print LOG_FILE "Converting String From : " . $each_line;
            $each_line =~ s/$str_to_replace/$str_replace_with/ig;
            print LOG_FILE "Converting String To   : " . $each_line;
            push(@newlines, $each_line);
            $files_changed{$file} = 1;
        } else {
            push(@newlines, $each_line);
            $files_not_changed{$file} = 1;
        }   
    }

    open(OUT_FILE, ">$output_dir$file") || warn "-W- Cant open file for writing: " . "$output_dir$file";
    print OUT_FILE @newlines;
    close(OUT_FILE);
}

my @f_changed     = keys %files_changed;
my @f_not_changed = keys %files_not_changed;

print LOG_FILE "\n\nOutput Summary as below ------------------------- : ";

print LOG_FILE "\n List of all the Files changed     : " . Dumper(\@f_changed);
print LOG_FILE "\n List of all the Files NOT changed : " . Dumper(\@f_not_changed);

print LOG_FILE "\n\n Total No.of files changed   : " . scalar(@f_changed);
print LOG_FILE "\n Total No.of files NOT changed : " . scalar(@f_not_changed);

close (LOG_FILE);
closedir(IN_DIR);
closedir(OUT_DIR);
closedir(LOG_DIR);

1;  

Input Directory: /usr/src/perl_test/input/

1) mother_teresa_intro.txt
Mother Teresa born on August 26, 1910
Full name of Mother Teresa is "Agnes Gonxha Bojaxhiu"
Mother Teresa founded the Missionaries of Charity, which in 2012 consisted of over 4,500 sisters and is active in 133 countries.

2) mother_teresa_awards.txt
In 1962, Mother Teresa was awarded the Ramon Magsaysay Award
In 1979, Mother Teresa was awarded the Nobel Peace Prize, for work undertaken in the struggle to overcome poverty and distress
In 1980, Mother Teresa was awarded the Bharat Ratna Prize  

Output Directory: /usr/src/perl_test/output/

1) mother_teresa_intro.txt
MOTHER TERESA born on August 26, 1910
Full name of MOTHER TERESA is "Agnes Gonxha Bojaxhiu"
MOTHER TERESA founded the Missionaries of Charity, which in 2012 consisted of over 4,500 sisters and is active in 133 countries.


2) mother_teresa_awards.txt
In 1962, MOTHER TERESA was awarded the Ramon Magsaysay Award
In 1979, MOTHER TERESA was awarded the Nobel Peace Prize, for work undertaken in the struggle to overcome poverty and distress
In 1980, MOTHER TERESA was awarded the Bharat Ratna Prize  

Log Directory: /usr/src/perl_test/log/
log_<time_stamp>.log