ਅਪਾਚੇ ਸਪਾਰਕ ਦੁਆਰਾ ਚਿੱਤਰ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਲਈ ਯੂਡੀਐਫ ਦੀ ਵਰਤੋਂ ਨਾਲ ਸਪਾਰਕਕੰਟੈਕਸ ਸਮੱਸਿਆਵਾਂ ਨੂੰ ਹੱਲ ਕਰਨਾ

ਅਪਾਚੇ ਸਪਾਰਕ ਦੇ UDF ਵਿੱਚ ਸਪਾਰਕਕੰਟੈਕਸ ਗਲਤੀਆਂ ਦੇ ਪਿੱਛੇ ਦੇ ਰਹੱਸ ਨੂੰ ਉਜਾਗਰ ਕਰਨਾ

ਨਾਲ ਕੰਮ ਕਰ ਰਿਹਾ ਹੈ ਅਪਾਚੇ ਸਪਾਰਕ ਅਤੇ ਪਾਈਸਪਾਰਕ ਵਿੱਚ ਅਕਸਰ ਵੱਡੇ ਪੈਮਾਨੇ ਦੇ ਡੇਟਾ ਕਾਰਜਾਂ ਨੂੰ ਸੰਭਾਲਣ ਲਈ ਵਿਤਰਿਤ ਕੰਪਿਊਟਿੰਗ ਦੀ ਵਰਤੋਂ ਸ਼ਾਮਲ ਹੁੰਦੀ ਹੈ। ਪਰ ਕਦੇ-ਕਦੇ, ਚੀਜ਼ਾਂ ਯੋਜਨਾ ਅਨੁਸਾਰ ਬਿਲਕੁਲ ਨਹੀਂ ਹੁੰਦੀਆਂ ਹਨ। ਇੱਕ ਆਮ ਸਮੱਸਿਆ ਬਹੁਤ ਸਾਰੇ ਡੇਟਾ ਵਿਗਿਆਨੀਆਂ ਨੂੰ ਆਉਂਦੀ ਹੈ, ਖਾਸ ਕਰਕੇ ਕਾਲ ਕਰਨ ਵੇਲੇ ਉਪਭੋਗਤਾ ਦੁਆਰਾ ਪਰਿਭਾਸ਼ਿਤ ਫੰਕਸ਼ਨ (UDFs), ਬਦਨਾਮ "ਸਪਾਰਕਕੰਟੈਕਸ ਸਿਰਫ ਡਰਾਈਵਰ 'ਤੇ ਵਰਤਿਆ ਜਾ ਸਕਦਾ ਹੈ" ਗਲਤੀ ਹੈ।

ਇਹ ਗਲਤੀ ਖਾਸ ਤੌਰ 'ਤੇ ਨਿਰਾਸ਼ਾਜਨਕ ਹੋ ਸਕਦੀ ਹੈ ਜਦੋਂ ਚਿੱਤਰ ਪ੍ਰੋਸੈਸਿੰਗ ਵਰਗੇ ਗੁੰਝਲਦਾਰ ਓਪਰੇਸ਼ਨ ਕਰਦੇ ਹਨ, ਜਿੱਥੇ ਕੰਮ ਕਈ ਕਰਮਚਾਰੀਆਂ ਵਿੱਚ ਵੰਡੇ ਜਾਂਦੇ ਹਨ। ਚਿੱਤਰ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਵਰਗੇ ਦ੍ਰਿਸ਼ਾਂ ਵਿੱਚ, ਇਹ ਸਮਝਣਾ ਕਿ SparkContext ਇਸ ਤਰ੍ਹਾਂ ਕਿਉਂ ਵਿਵਹਾਰ ਕਰਦਾ ਹੈ ਮਹੱਤਵਪੂਰਨ ਬਣ ਜਾਂਦਾ ਹੈ। 💻

ਇਸ ਲੇਖ ਵਿੱਚ, ਮੈਂ ਤੁਹਾਨੂੰ ਪਾਈਟੋਰਚ ਵਿੱਚ ResNet ਮਾਡਲ ਨੂੰ ਸ਼ਾਮਲ ਕਰਨ ਵਾਲੀ ਇੱਕ ਉਦਾਹਰਣ ਦੇ ਕੇ ਲੈ ਜਾਵਾਂਗਾ। ਅਸੀਂ ਖੋਜ ਕਰਾਂਗੇ ਕਿ UDF ਦੇ ਅੰਦਰ ਓਪਰੇਸ਼ਨਾਂ ਨੂੰ ਲੜੀਬੱਧ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਦੇ ਸਮੇਂ SparkContext ਸਮੱਸਿਆਵਾਂ ਕਿਉਂ ਪੈਦਾ ਕਰਦਾ ਹੈ, ਜਿਸ ਨਾਲ ਰਨਟਾਈਮ ਗਲਤੀ ਹੁੰਦੀ ਹੈ। ਇਸਦੇ ਦੁਆਰਾ, ਮੈਂ ਸਪਾਰਕ ਦੇ ਨਾਲ ਨਿਰਵਿਘਨ ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਨੂੰ ਸਮਰੱਥ ਕਰਨ ਲਈ ਗਲਤੀ ਦੇ ਆਲੇ-ਦੁਆਲੇ ਕੰਮ ਕਰਨ ਲਈ ਰਣਨੀਤੀਆਂ ਵੀ ਸਾਂਝੀਆਂ ਕਰਾਂਗਾ।

ਜੇ ਤੁਸੀਂ ਸਪਾਰਕ ਵਿੱਚ ਇੱਕ ਐਮਐਲ ਪਾਈਪਲਾਈਨ ਬਣਾਉਂਦੇ ਸਮੇਂ ਇਸ ਮੁੱਦੇ ਦਾ ਸਾਹਮਣਾ ਕੀਤਾ ਹੈ, ਤਾਂ ਤੁਸੀਂ ਇਕੱਲੇ ਨਹੀਂ ਹੋ! ਮੇਰੇ ਨਾਲ ਰਹੋ ਕਿਉਂਕਿ ਅਸੀਂ ਇਸ ਗਲਤੀ ਤੋਂ ਬਚਣ ਅਤੇ ਵਿਤਰਿਤ ਵਾਤਾਵਰਣਾਂ ਵਿੱਚ ਸਪਾਰਕ UDFs ਦੇ ਸੁਚਾਰੂ ਸੰਚਾਲਨ ਨੂੰ ਯਕੀਨੀ ਬਣਾਉਣ ਲਈ ਵਿਹਾਰਕ ਹੱਲ ਲੱਭਦੇ ਹਾਂ। 🚀

ਡੀਪ ਲਰਨਿੰਗ ਮਾਡਲਾਂ ਨਾਲ ਸਪਾਰਕ UDF ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਦਾ ਨਿਪਟਾਰਾ ਕਰਨਾ

ਨਾਲ ਕੰਮ ਕਰਦੇ ਸਮੇਂ ਅਪਾਚੇ ਸਪਾਰਕ, ਡਿਸਟ੍ਰੀਬਿਊਟਿਡ ਪ੍ਰੋਸੈਸਿੰਗ ਅਕਸਰ ਓਪਰੇਸ਼ਨਾਂ ਨੂੰ ਤੇਜ਼ ਕਰਨ ਲਈ ਵਰਤੀ ਜਾਂਦੀ ਹੈ, ਖਾਸ ਤੌਰ 'ਤੇ ਵੱਡੇ ਪੈਮਾਨੇ ਦੀ ਚਿੱਤਰ ਪ੍ਰੋਸੈਸਿੰਗ ਵਰਗੇ ਕੰਮਾਂ ਵਿੱਚ। ਹਾਲਾਂਕਿ, ਸਪਾਰਕ ਕੁਝ ਪਾਬੰਦੀਆਂ ਲਾਉਂਦਾ ਹੈ, ਖਾਸ ਤੌਰ 'ਤੇ ਇਸ 'ਤੇ ਸਪਾਰਕ ਸੰਦਰਭ. ਉਪਰੋਕਤ ਸਕ੍ਰਿਪਟਾਂ ਵਿੱਚ, ਇੱਕ ਡੇਟਾਫ੍ਰੇਮ ਵਿੱਚ ਹਰੇਕ ਕਤਾਰ ਲਈ ਚਿੱਤਰਾਂ ਤੋਂ ਵਿਸ਼ੇਸ਼ਤਾਵਾਂ ਨੂੰ ਐਕਸਟਰੈਕਟ ਕਰਨ ਲਈ ਇੱਕ UDF ਦੇ ਅੰਦਰ ResNet ਡੂੰਘੇ ਸਿਖਲਾਈ ਮਾਡਲ ਦੀ ਵਰਤੋਂ ਕੀਤੀ ਜਾਂਦੀ ਹੈ। ਇਹ ਪਹੁੰਚ ਇੱਕ SparkContext ਸੀਮਾ ਨੂੰ ਹਿੱਟ ਕਰਦਾ ਹੈ: SparkContext ਦੀ ਵਰਤੋਂ ਸਿਰਫ਼ ਡਰਾਈਵਰ ਨੋਡ 'ਤੇ ਕੀਤੀ ਜਾ ਸਕਦੀ ਹੈ ਨਾ ਕਿ ਵਰਕਰ ਨੋਡਾਂ 'ਤੇ ਚੱਲ ਰਹੇ ਕੋਡ ਦੇ ਅੰਦਰ, ਜਿਸ ਕਰਕੇ ਕੋਡ ਇੱਕ ਗਲਤੀ ਸੁੱਟਦਾ ਹੈ। ਸ਼ੁਰੂਆਤੀ ਹੱਲ ਵਿੱਚ ਸਪਾਰਕ ਸੈਸ਼ਨ, ਚਿੱਤਰ ਪ੍ਰੀ-ਪ੍ਰੋਸੈਸਿੰਗ, ਅਤੇ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਲਈ ਇੱਕ ਇਮੇਜਵੈਕਟਰਾਈਜ਼ਰ ਕਲਾਸ ਬਣਾਉਣਾ ਸ਼ਾਮਲ ਹੈ। ਇਹਨਾਂ ਕੰਮਾਂ ਨੂੰ ਇੱਕ ਕਲਾਸ ਵਿੱਚ ਕੇਂਦਰਿਤ ਕਰਕੇ, ਅਸੀਂ ਕੋਡ ਨੂੰ ਮਾਡਿਊਲਰ ਅਤੇ ਅਨੁਕੂਲ ਰੱਖਣ ਦੇ ਯੋਗ ਹੁੰਦੇ ਹਾਂ। 💻

ਪਹਿਲੀ ਸਕ੍ਰਿਪਟ ਵਿੱਚ, ਇਮੇਜਵੈਕਟੋਰਾਈਜ਼ਰ ਕਲਾਸ ਇੱਕ ਸਪਾਰਕ ਸੈਸ਼ਨ ਸ਼ੁਰੂ ਕਰਦਾ ਹੈ ਅਤੇ ਇੱਕ ਪ੍ਰਸਿੱਧ ਡੂੰਘੀ ਸਿਖਲਾਈ ਲਾਇਬ੍ਰੇਰੀ, PyTorch ਤੋਂ ਇੱਕ ਪ੍ਰੀ-ਟ੍ਰੇਂਡ ResNet ਮਾਡਲ ਲੋਡ ਕਰਦਾ ਹੈ। ਰੀਸਾਈਜ਼ਿੰਗ ਅਤੇ ਸਧਾਰਣ ਬਣਾਉਣ ਸਮੇਤ, ਲਾਗੂ ਕੀਤੇ ਪਰਿਵਰਤਨਾਂ ਦੇ ਇੱਕ ਸਮੂਹ ਦੇ ਨਾਲ, ਹਰੇਕ ਚਿੱਤਰ ਨੂੰ ਮਾਡਲ ਲਈ ਇੱਕ ਅਨੁਕੂਲ ਫਾਰਮੈਟ ਵਿੱਚ ਬਦਲਿਆ ਜਾ ਸਕਦਾ ਹੈ। extract_features ਵਿਧੀ ਇਹ ਪਰਿਭਾਸ਼ਿਤ ਕਰਦੀ ਹੈ ਕਿ ਹਰੇਕ ਚਿੱਤਰ ਨੂੰ ਕਿਵੇਂ ਸੰਸਾਧਿਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ: ਪਹਿਲਾਂ, ਚਿੱਤਰ ਨੂੰ ਪੜ੍ਹਿਆ ਜਾਂਦਾ ਹੈ, ਪ੍ਰੀ-ਪ੍ਰੋਸੈਸ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਫਿਰ ਉੱਚ-ਪੱਧਰੀ ਵਿਸ਼ੇਸ਼ਤਾ ਵੈਕਟਰਾਂ ਨੂੰ ਐਕਸਟਰੈਕਟ ਕਰਨ ਲਈ ResNet ਮਾਡਲ ਦੁਆਰਾ ਪਾਸ ਕੀਤਾ ਜਾਂਦਾ ਹੈ। ਹਾਲਾਂਕਿ, ਇਹ ਪਹੁੰਚ SparkContext ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਮੁੱਦੇ ਨੂੰ ਹਿੱਟ ਕਰਦੀ ਹੈ ਕਿਉਂਕਿ UDF ਸਪਾਰਕ ਕੰਪੋਨੈਂਟਸ ਨੂੰ ਸਿੱਧੇ ਵਰਕਰ ਦੇ ਕੰਮਾਂ ਦੇ ਅੰਦਰ ਐਕਸੈਸ ਕਰਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਦਾ ਹੈ। ਕਿਉਂਕਿ PySpark ਵੰਡੇ ਹੋਏ ਨੋਡਾਂ 'ਤੇ ਚੱਲਣ ਲਈ ResNet ਮਾਡਲ ਨੂੰ ਸੀਰੀਅਲਾਈਜ਼ ਨਹੀਂ ਕਰ ਸਕਦਾ, ਇਹ ਰਨਟਾਈਮ ਮੁੱਦਾ ਬਣਾਉਂਦਾ ਹੈ।

ਇਸ ਨੂੰ ਹੱਲ ਕਰਨ ਲਈ, ਦੂਜੀ ਪਹੁੰਚ ਸਪਾਰਕ ਦੀ ਵਰਤੋਂ ਕਰਦੀ ਹੈ ਪ੍ਰਸਾਰਣ ਵੇਰੀਏਬਲ, ਜੋ ਹਰੇਕ ਵਰਕਰ ਨੂੰ ਸਿਰਫ਼ ਇੱਕ ਵਾਰ ਡਾਟਾ ਜਾਂ ਵਸਤੂਆਂ ਵੰਡਦੇ ਹਨ। ResNet ਮਾਡਲ ਦਾ ਪ੍ਰਸਾਰਣ ਮਾਡਲ ਨੂੰ ਹਰੇਕ ਵਰਕਰ ਨੋਡ 'ਤੇ ਸਟੋਰ ਕਰਨ ਦੀ ਇਜਾਜ਼ਤ ਦਿੰਦਾ ਹੈ ਅਤੇ ਹਰੇਕ UDF ਕਾਲ ਵਿੱਚ ਮੁੜ-ਸ਼ੁਰੂ ਹੋਣ ਤੋਂ ਰੋਕਦਾ ਹੈ। ਪ੍ਰਸਾਰਣ ਮਾਡਲ ਨੂੰ ਚਿੱਤਰ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਦੇ ਦੌਰਾਨ ਹਵਾਲਾ ਦਿੱਤਾ ਜਾਂਦਾ ਹੈ, ਸੈੱਟਅੱਪ ਨੂੰ ਵਧੇਰੇ ਕੁਸ਼ਲ ਅਤੇ ਸਕੇਲੇਬਲ ਬਣਾਉਂਦਾ ਹੈ। ਇਹ ਵਿਧੀ ਸਰੋਤਾਂ ਦੀ ਵਰਤੋਂ ਨੂੰ ਮਹੱਤਵਪੂਰਨ ਤੌਰ 'ਤੇ ਘਟਾਉਂਦੀ ਹੈ ਅਤੇ ਇਹ ਯਕੀਨੀ ਬਣਾ ਕੇ ਸਪਾਰਕਕੰਟੈਕਸ ਗਲਤੀ ਤੋਂ ਬਚਦੀ ਹੈ ਕਿ ਸਪਾਰਕ ਸਿਰਫ ਡਰਾਈਵਰ 'ਤੇ ਲੋੜੀਂਦੇ ਭਾਗਾਂ ਤੱਕ ਪਹੁੰਚ ਕਰਦਾ ਹੈ, ਕਰਮਚਾਰੀਆਂ 'ਤੇ ਨਹੀਂ। ਬਰਾਡਕਾਸਟ ਵੇਰੀਏਬਲ ਵਿਸ਼ੇਸ਼ ਤੌਰ 'ਤੇ ਉਪਯੋਗੀ ਹੁੰਦੇ ਹਨ ਜਦੋਂ ਸਮਾਨਾਂਤਰ ਵਿੱਚ ਵੱਡੇ ਡੇਟਾਸੇਟਾਂ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰਦੇ ਹੋਏ, ਦੂਜੀ ਸਕ੍ਰਿਪਟ ਨੂੰ ਵਿਤਰਿਤ ਚਿੱਤਰ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਲਈ ਆਦਰਸ਼ ਬਣਾਉਂਦੇ ਹਨ।

ਬ੍ਰੌਡਕਾਸਟ ਮਾਡਲ ਦੀ ਵਰਤੋਂ ਕਰਨ ਲਈ UDF ਫੰਕਸ਼ਨ ਨੂੰ ਐਡਜਸਟ ਕਰਨ ਤੋਂ ਬਾਅਦ, ਅਸੀਂ ਇੱਕ UDF ਨੂੰ ਪਰਿਭਾਸ਼ਿਤ ਕਰਦੇ ਹਾਂ ਜੋ ਡੇਟਾਫ੍ਰੇਮ ਦੀ ਹਰੇਕ ਕਤਾਰ 'ਤੇ ਪਰਿਵਰਤਨ ਲਾਗੂ ਕਰਦਾ ਹੈ। ਇਹ ਪੁਸ਼ਟੀ ਕਰਨ ਲਈ ਕਿ ਸਕ੍ਰਿਪਟਾਂ ਵੱਖ-ਵੱਖ ਵਾਤਾਵਰਣਾਂ ਵਿੱਚ ਕੰਮ ਕਰਦੀਆਂ ਹਨ, ਇੱਕ ਤੀਜੀ ਸਕ੍ਰਿਪਟ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਯੂਨਿਟ ਟੈਸਟਿੰਗ ਲਈ ਪ੍ਰਦਾਨ ਕੀਤੀ ਜਾਂਦੀ ਹੈ ਪਾਈਟੈਸਟ. ਇਹ ਸਕ੍ਰਿਪਟ ਬਾਈਨਰੀ ਚਿੱਤਰ ਡੇਟਾ ਨੂੰ ਸੰਭਾਲਣ, ਪਰਿਵਰਤਨ ਪਾਈਪਲਾਈਨ ਨੂੰ ਚਲਾਉਣ, ਅਤੇ ਇੱਕ ਸਹੀ ਆਕਾਰ ਦੇ ਵਿਸ਼ੇਸ਼ਤਾ ਵੈਕਟਰ ਨੂੰ ਆਉਟਪੁੱਟ ਕਰਨ ਲਈ ਫੰਕਸ਼ਨ ਦੀ ਯੋਗਤਾ ਦੀ ਜਾਂਚ ਕਰਦੀ ਹੈ। ਜਾਂਚ ਤੈਨਾਤੀ ਤੋਂ ਪਹਿਲਾਂ ਹਰੇਕ ਹਿੱਸੇ ਦੇ ਫੰਕਸ਼ਨ ਦੀ ਪੁਸ਼ਟੀ ਕਰਕੇ ਭਰੋਸੇਯੋਗਤਾ ਦੀ ਇੱਕ ਹੋਰ ਪਰਤ ਜੋੜਦੀ ਹੈ। 📊 ਯੂਨਿਟ ਟੈਸਟ ਵਿਤਰਿਤ ਵਾਤਾਵਰਣਾਂ ਵਿੱਚ ਵਿਸ਼ੇਸ਼ ਤੌਰ 'ਤੇ ਕੀਮਤੀ ਹੁੰਦੇ ਹਨ, ਕਿਉਂਕਿ ਉਹ ਇਹ ਯਕੀਨੀ ਬਣਾਉਂਦੇ ਹਨ ਕਿ ਕੋਡ ਸੋਧਾਂ ਨੋਡਾਂ ਵਿੱਚ ਅਣਇੱਛਤ ਸਮੱਸਿਆਵਾਂ ਪੇਸ਼ ਨਹੀਂ ਕਰਦੀਆਂ ਹਨ।

ਅਸਲ-ਸੰਸਾਰ ਦੀਆਂ ਐਪਲੀਕੇਸ਼ਨਾਂ ਵਿੱਚ, ਇਹ ਪਹੁੰਚ ਸਮਾਨਾਂਤਰ ਵਿੱਚ ਗੁੰਝਲਦਾਰ ਚਿੱਤਰ ਡੇਟਾ ਨੂੰ ਸੰਭਾਲਣ ਦੀ ਸਪਾਰਕ ਦੀ ਯੋਗਤਾ ਨੂੰ ਵਧਾਉਂਦੇ ਹਨ, ਜਿਸ ਨਾਲ ਮਸ਼ੀਨ ਸਿਖਲਾਈ ਅਤੇ AI ਪ੍ਰੋਜੈਕਟਾਂ ਵਿੱਚ ਵਿਸ਼ਾਲ ਚਿੱਤਰ ਡੇਟਾਸੈਟਾਂ ਨਾਲ ਕੰਮ ਕਰਨਾ ਸੰਭਵ ਹੋ ਜਾਂਦਾ ਹੈ। ਬ੍ਰੌਡਕਾਸਟ ਮਾਡਲ, UDF, ਅਤੇ ਟੈਸਟਿੰਗ ਫਰੇਮਵਰਕ ਇਹਨਾਂ ਵਰਕਫਲੋਜ਼ ਨੂੰ ਅਨੁਕੂਲ ਬਣਾਉਣ ਵਿੱਚ ਮਹੱਤਵਪੂਰਣ ਭੂਮਿਕਾਵਾਂ ਨਿਭਾਉਂਦੇ ਹਨ। ਇਹ ਹੱਲ ਵੱਡੇ ਪੈਮਾਨੇ ਦੀ ਡਾਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਲਈ ਲਚਕਤਾ, ਮਾਪਯੋਗਤਾ ਅਤੇ ਭਰੋਸੇਯੋਗਤਾ ਲਿਆਉਂਦੇ ਹਨ—ਵਿਤਰਿਤ ਮਸ਼ੀਨ ਸਿਖਲਾਈ ਪਾਈਪਲਾਈਨਾਂ ਵਿੱਚ ਇਕਸਾਰ, ਉੱਚ-ਗੁਣਵੱਤਾ ਦੇ ਨਤੀਜੇ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਮਹੱਤਵਪੂਰਨ।

# Import required libraries
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, FloatType
from torchvision import models, transforms
from PIL import Image
import torch
import numpy as np
from io import BytesIO
# Define the class to initialize Spark session and ResNet model
class ImageVectorizer:
    def __init__(self):
        # Initialize SparkSession
        self.spark = SparkSession.builder.getOrCreate()
        # Load pre-trained ResNet model
        self.resnet_model = models.resnet50(pretrained=True)
        self.resnet_model.eval()
        # Define image transformation pipeline
        self.transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225])
        ])
    def extract_features(self, image_binary):
        # Convert image binary to tensor and extract features
        image = Image.open(BytesIO(image_binary))
        image = self.transform(image).unsqueeze(0)
        with torch.no_grad():
            features = self.resnet_model(image)
        return features.squeeze().numpy().tolist()
    def process_images(self, image_df):
        # Register a non-Spark UDF to call extract_features function
        extract_features_udf = udf(lambda x: self.extract_features(x), ArrayType(FloatType()))
        return image_df.withColumn("features", extract_features_udf(image_df["content"]))

# Import required libraries
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, FloatType
from torchvision import models, transforms
from PIL import Image
import torch
import numpy as np
from io import BytesIO
# Initialize Spark session and broadcast model
spark = SparkSession.builder.getOrCreate()
resnet_model = models.resnet50(pretrained=True)
resnet_model.eval()
bc_resnet_model = spark.sparkContext.broadcast(resnet_model)
# Define transformation pipeline separately
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                     std=[0.229, 0.224, 0.225])
])
# Define feature extraction function using broadcast model
def extract_features(image_binary):
    image = Image.open(BytesIO(image_binary))
    image = transform(image).unsqueeze(0)
    with torch.no_grad():
        features = bc_resnet_model.value(image)
    return features.squeeze().numpy().tolist()
# Register UDF
extract_features_udf = udf(extract_features, ArrayType(FloatType()))

# Import pytest for unit testing
import pytest
import numpy as np
@pytest.fixture
def mock_image_binary():
    # Provide a sample image in binary format
    with open('test_image.jpg', 'rb') as f:
        return f.read()
def test_extract_features(mock_image_binary):
    # Initialize ImageVectorizer and call extract_features function
    vectorizer = ImageVectorizer()
    result = vectorizer.extract_features(mock_image_binary)
    assert isinstance(result, list)
    assert len(result) == 2048

ਚਿੱਤਰ ਪ੍ਰੋਸੈਸਿੰਗ ਲਈ ਸਪਾਰਕ UDF ਦੇ ਨਾਲ ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਚੁਣੌਤੀਆਂ ਨੂੰ ਪਾਰ ਕਰਨਾ

ਵਰਤਣ ਵਿੱਚ ਮਹੱਤਵਪੂਰਨ ਚੁਣੌਤੀਆਂ ਵਿੱਚੋਂ ਇੱਕ ਅਪਾਚੇ ਸਪਾਰਕ ਜਿਵੇਂ ਕਿ ਉੱਨਤ ਕੰਮਾਂ ਲਈ ਚਿੱਤਰ ਨੂੰ ਕਾਰਵਾਈ ਕਰਨ ਉਪਭੋਗਤਾ-ਪ੍ਰਭਾਸ਼ਿਤ ਫੰਕਸ਼ਨਾਂ (UDFs) ਨਾਲ ਕੰਮ ਕਰਦੇ ਸਮੇਂ ਨਿਰਵਿਘਨ ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਨੂੰ ਯਕੀਨੀ ਬਣਾ ਰਿਹਾ ਹੈ। ਕਿਉਂਕਿ ਸਪਾਰਕ ਕੁਦਰਤੀ ਤੌਰ 'ਤੇ ਵੰਡਿਆ ਜਾਂਦਾ ਹੈ, ਸਪਾਰਕ UDFs ਦੇ ਅੰਦਰ ਕਾਰਜਾਂ ਨੂੰ ਪ੍ਰੋਸੈਸਿੰਗ ਲਈ ਵਰਕਰ ਨੋਡਾਂ ਨੂੰ ਭੇਜਿਆ ਜਾਂਦਾ ਹੈ, ਜੋ ਸਮੱਸਿਆਵਾਂ ਪੈਦਾ ਕਰ ਸਕਦੇ ਹਨ ਜੇਕਰ ਗੈਰ-ਸੀਰੀਅਲਾਈਜ਼ਯੋਗ ਵਸਤੂਆਂ ਜਿਵੇਂ ਕਿ ਗੁੰਝਲਦਾਰ ਮਸ਼ੀਨ ਸਿਖਲਾਈ ਮਾਡਲ ਸ਼ਾਮਲ ਹੁੰਦੇ ਹਨ। PyTorch ਤੋਂ ResNet ਮਾਡਲ, ਉਦਾਹਰਨ ਲਈ, ਨੇਟਿਵ ਤੌਰ 'ਤੇ ਸੀਰੀਅਲਾਈਜ਼ ਕਰਨ ਯੋਗ ਨਹੀਂ ਹੈ, ਮਤਲਬ ਕਿ ਇਸਨੂੰ "ਸਪਾਰਕ ਕਨਟੈਕਸਟ ਸਿਰਫ ਡਰਾਈਵਰ 'ਤੇ ਵਰਤਿਆ ਜਾ ਸਕਦਾ ਹੈ" ਗਲਤੀ ਤੋਂ ਬਚਣ ਲਈ ਸਪਾਰਕ ਦੇ ਅੰਦਰ ਧਿਆਨ ਨਾਲ ਹੈਂਡਲ ਕਰਨ ਦੀ ਲੋੜ ਹੈ।

ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਇੱਕ ਰੁਕਾਵਟ ਬਣ ਜਾਂਦੀ ਹੈ ਕਿਉਂਕਿ ਸਪਾਰਕ UDF ਵਿੱਚ ਹਵਾਲਾ ਦਿੱਤੇ ਸਾਰੇ ਤੱਤਾਂ ਨੂੰ ਵੰਡਣ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਦਾ ਹੈ, SparkContext ਸਮੇਤ, ਸਿੱਧੇ ਵਰਕਰ ਨੋਡਾਂ ਵਿੱਚ। ਇਹ ਸੀਮਾ ਇਸ ਲਈ ਹੈ ਕਿ ਅਸੀਂ ਹਰ ਵਾਰ ਮੁੜ-ਸ਼ੁਰੂ ਕੀਤੇ ਬਿਨਾਂ ResNet ਮਾਡਲ ਨੂੰ ਨੋਡਾਂ ਵਿੱਚ ਕੁਸ਼ਲਤਾ ਨਾਲ ਸਾਂਝਾ ਕਰਨ ਲਈ ਇੱਕ ਪ੍ਰਸਾਰਣ ਵੇਰੀਏਬਲ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹਾਂ। ਅਜਿਹੇ ਮਾਮਲਿਆਂ ਵਿੱਚ, ਦ broadcast() ਵਿਧੀ ਹਰੇਕ ਕਰਮਚਾਰੀ ਨੂੰ ਸਿਰਫ਼-ਪੜ੍ਹਨ ਵਾਲੇ ਡੇਟਾ ਨੂੰ ਵੰਡਣ ਵਿੱਚ ਮਦਦ ਕਰਦੀ ਹੈ, ਜਿੱਥੇ ਸਪਾਰਕ ਦੀਆਂ ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਪਾਬੰਦੀਆਂ ਨੂੰ ਚਾਲੂ ਕੀਤੇ ਬਿਨਾਂ ਇਸਨੂੰ ਸਥਾਨਕ ਤੌਰ 'ਤੇ ਹਵਾਲਾ ਦਿੱਤਾ ਜਾ ਸਕਦਾ ਹੈ। ਮਾਡਲ ਨੂੰ ਪ੍ਰਸਾਰਿਤ ਕਰਨ ਦੁਆਰਾ, ਰੀਸਨੈੱਟ ਵੇਟ ਸਾਰੇ ਨੋਡਾਂ 'ਤੇ ਡੇਟਾ ਨੂੰ ਡੁਪਲੀਕੇਟ ਕੀਤੇ ਬਿਨਾਂ ਵਿਸ਼ੇਸ਼ਤਾ ਕੱਢਣ ਲਈ ਪਹੁੰਚਯੋਗ ਹੁੰਦੇ ਹਨ, ਮੈਮੋਰੀ ਵਰਤੋਂ ਅਤੇ ਪ੍ਰਦਰਸ਼ਨ ਦੋਵਾਂ ਨੂੰ ਵਧਾਉਂਦੇ ਹਨ। 🌍

ਇਹ ਤਕਨੀਕ ਚਿੱਤਰ ਪ੍ਰੋਸੈਸਿੰਗ ਤੋਂ ਪਰੇ ਵੰਡੀਆਂ ML ਪਾਈਪਲਾਈਨਾਂ ਲਈ ਵਿਆਪਕ ਤੌਰ 'ਤੇ ਲਾਗੂ ਹੁੰਦੀ ਹੈ। ਉਦਾਹਰਨ ਲਈ, ਜੇਕਰ ਤੁਸੀਂ ਇੱਕ ਸਿਫ਼ਾਰਿਸ਼ ਪ੍ਰਣਾਲੀ ਨੂੰ ਲਾਗੂ ਕਰ ਰਹੇ ਹੋ, ਤਾਂ ਤੁਸੀਂ ਸਪਾਰਕ ਸੀਰੀਅਲਾਈਜ਼ੇਸ਼ਨ ਤਰੁਟੀਆਂ ਤੋਂ ਬਚਣ ਲਈ ਉਪਭੋਗਤਾ ਤਰਜੀਹਾਂ ਜਾਂ ਪ੍ਰੀ-ਟ੍ਰੇਂਡ ਮਾਡਲਾਂ ਦੇ ਵੱਡੇ ਡੇਟਾਸੈਟਾਂ ਨੂੰ ਪ੍ਰਸਾਰਿਤ ਕਰ ਸਕਦੇ ਹੋ। ਇਸੇ ਤਰ੍ਹਾਂ, ਹੋਰ ਪ੍ਰੀ-ਪ੍ਰੋਸੈਸਿੰਗ ਕਾਰਜਾਂ (ਜਿਵੇਂ ਕਿ ਟੈਕਸਟ ਵੈਕਟੋਰਾਈਜ਼ੇਸ਼ਨ ਜਾਂ ਆਡੀਓ ਪ੍ਰੋਸੈਸਿੰਗ) ਲਈ UDFs ਦੀ ਵਰਤੋਂ ਕਰਨ ਨਾਲ ਗੈਰ-ਸੀਰੀਅਲਾਈਜ਼ ਕਰਨ ਯੋਗ ਵਸਤੂਆਂ ਦੇ ਪ੍ਰਸਾਰਣ ਤੋਂ ਵੀ ਲਾਭ ਹੁੰਦਾ ਹੈ, ਜਿਸ ਨਾਲ ਸਪਾਰਕ ਨੂੰ ਡਾਟਾ ਡੁਪਲੀਕੇਸ਼ਨ ਓਵਰਹੈੱਡਸ ਤੋਂ ਬਿਨਾਂ ਉੱਚੇ ਸਮਾਨਾਂਤਰ ਕਾਰਜਾਂ ਨੂੰ ਸੰਭਾਲਣ ਦੀ ਆਗਿਆ ਮਿਲਦੀ ਹੈ। ਇਹ ਅਭਿਆਸ ਸਪਾਰਕ ਨੂੰ ਵਧੀਆ ML ਵਰਕਫਲੋ ਨੂੰ ਸੰਭਾਲਣ ਲਈ ਕਾਫ਼ੀ ਮਜ਼ਬੂਤ ​​ਬਣਾਉਂਦੇ ਹਨ, ਜਿਸ ਨਾਲ ਢਾਂਚਾਗਤ ਅਤੇ ਗੈਰ-ਸੰਗਠਿਤ ਡੇਟਾ ਕਾਰਜਾਂ ਵਿੱਚ ਵੱਡੇ ਡੇਟਾਸੇਟਾਂ ਲਈ ਲੋੜੀਂਦੀ ਮਾਪਯੋਗਤਾ ਪ੍ਰਦਾਨ ਕੀਤੀ ਜਾਂਦੀ ਹੈ। 🚀